Content

Content

DomoJupyter_Content dataclass

DomoJupyter_Content(
    name: str,
    folder: str,
    last_modified: datetime,
    file_type: str,
    content: str,
    auth: DomoJupyterAuth,
    dj_workspace: Any = None,
    target_folder: str = "export",
)

Bases: DomoBase

export

export(
    *,
    target_parent: str | None = None,
    file_name: str | None = None,
    target_folder: str | None = None
) -> str

Write this item's content under a local path.

Parameters:

Name	Type	Description	Default
target_parent	str | None	Full path to the directory that will contain the file (if set, self.folder is ignored for layout).	None
file_name	str | None	Output file name (default: self.name).	None
target_folder	str | None	If set, updates instance target_folder then uses os.path.join(self.target_folder, self.folder) as the parent directory (unless target_parent is set).	None

Source code in src/crew_dcs/classes/DomoJupyter/Content.py

def export(
    self,
    *,
    target_parent: str | None = None,
    file_name: str | None = None,
    target_folder: str | None = None,
) -> str:
    """Write this item's content under a local path.

    Args:
        target_parent: Full path to the directory that will contain the file
            (if set, ``self.folder`` is ignored for layout).
        file_name: Output file name (default: ``self.name``).
        target_folder: If set, updates instance ``target_folder`` then uses
            ``os.path.join(self.target_folder, self.folder)`` as the parent
            directory (unless ``target_parent`` is set).
    """
    if target_folder is not None:
        self.target_folder = target_folder

    parent_dir = target_parent or os.path.join(self.target_folder, self.folder)

    out_name = file_name or self.name

    if not os.path.exists(parent_dir):
        logger.debug("Creating output folder: %s", parent_dir)
        os.makedirs(parent_dir)

    content_str = self.content
    if isinstance(self.content, dict):
        content_str = json.dumps(self.content)

    output_path = os.path.join(parent_dir, out_name)
    with open(output_path, "w", encoding="utf-8") as f:
        f.write(content_str)

    return output_path

write_under_target

write_under_target(
    *, target_folder: str, relative_path: str
) -> str

Write content to target_folder / relative_path (POSIX segments). Creates parents.

Source code in src/crew_dcs/classes/DomoJupyter/Content.py

def write_under_target(self, *, target_folder: str, relative_path: str) -> str:
    """Write content to ``target_folder`` / ``relative_path`` (POSIX segments). Creates parents."""
    rel = (relative_path or "").strip().replace("\\", "/").strip("/")
    if not rel:
        segments: list[str] = [self.name]
    else:
        segments = [p for p in rel.split("/") if p]
    out_path = os.path.join(target_folder, *segments)
    parent = os.path.dirname(out_path)
    if parent and not os.path.exists(parent):
        os.makedirs(parent, exist_ok=True)

    content_str = self.content
    if isinstance(self.content, dict):
        content_str = json.dumps(self.content)

    with open(out_path, "w", encoding="utf-8") as f:
        f.write(content_str)

    return out_path

JupyterNotebook dataclass

JupyterNotebook(
    name: str,
    folder: str,
    last_modified: datetime,
    file_type: str,
    content: str,
    auth: DomoJupyterAuth,
    dj_workspace: Any = None,
    target_folder: str = "export",
    notebook_id: str | None = None,
)

Bases: DomoJupyter_Content

Notebook-specific content with export, update from string/file, and dataflow (schedule) support.

Use JupyterNotebook.from_content(content, notebook_id=...) to wrap a DomoJupyter_Content when file_type == "notebook". notebook_id is required for get_dataflow/update_dataflow; it is the UUID used by the dataflow API (e.g. from Domo UI or browser network tab).

from_content classmethod

from_content(
    content: DomoJupyter_Content,
    notebook_id: str | None = None,
) -> JupyterNotebook

Wrap a content item as JupyterNotebook. Raises if content is not a notebook.

Source code in src/crew_dcs/classes/DomoJupyter/Content.py

@classmethod
def from_content(
    cls,
    content: DomoJupyter_Content,
    notebook_id: str | None = None,
) -> "JupyterNotebook":
    """Wrap a content item as JupyterNotebook. Raises if content is not a notebook."""
    if content.file_type != "notebook" and not content.name.endswith(".ipynb"):
        raise ValueError(
            f"Content is not a notebook (file_type={content.file_type!r}, name={content.name!r})"
        )
    return cls(
        name=content.name,
        folder=content.folder,
        last_modified=content.last_modified,
        file_type=content.file_type,
        content=content.content,
        auth=content.auth,
        dj_workspace=content.dj_workspace,
        target_folder=content.target_folder,
        notebook_id=notebook_id,
    )

get_dataflow async

get_dataflow(
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Get dataflow (scheduling) config for this notebook.

Requires dj_workspace and notebook_id. The Contents API does not expose notebook_id; it can be obtained from the Domo UI or browser network tab until the API provides it.

Source code in src/crew_dcs/classes/DomoJupyter/Content.py

async def get_dataflow(
    self,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Get dataflow (scheduling) config for this notebook.

    Requires dj_workspace and notebook_id. The Contents API does not expose
    notebook_id; it can be obtained from the Domo UI or browser network tab
    until the API provides it.
    """
    ws = self._require_dataflow_context()
    return await ws.get_notebook_dataflow(
        self.notebook_id,
        return_raw=return_raw,
        context=context,
        **context_kwargs,
    )

schedule_dataflow async

schedule_dataflow(
    body: dict[str, Any],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Alias for update_dataflow (set schedule / dataflow config).

Source code in src/crew_dcs/classes/DomoJupyter/Content.py

async def schedule_dataflow(
    self,
    body: dict[str, Any],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Alias for update_dataflow (set schedule / dataflow config)."""
    return await self.update_dataflow(
        body=body,
        return_raw=return_raw,
        context=context,
        **context_kwargs,
    )

update_dataflow async

update_dataflow(
    body: dict[str, Any],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Update dataflow (scheduling) config for this notebook.

Requires dj_workspace and notebook_id. notebook_id can be obtained from the Domo UI or browser (e.g. network tab) until the API exposes it.

Source code in src/crew_dcs/classes/DomoJupyter/Content.py

async def update_dataflow(
    self,
    body: dict[str, Any],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Update dataflow (scheduling) config for this notebook.

    Requires dj_workspace and notebook_id. notebook_id can be obtained from
    the Domo UI or browser (e.g. network tab) until the API exposes it.
    """
    ws = self._require_dataflow_context()
    return await ws.update_notebook_dataflow(
        self.notebook_id,
        body=body,
        return_raw=return_raw,
        context=context,
        **context_kwargs,
    )

update_from_file async

update_from_file(
    path: str | PathLike,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Read file and update notebook content. .ipynb loaded as JSON, else text.

Source code in src/crew_dcs/classes/DomoJupyter/Content.py

async def update_from_file(
    self,
    path: str | os.PathLike,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Read file and update notebook content. .ipynb loaded as JSON, else text."""
    path = os.fspath(path)
    with open(path, encoding="utf-8") as f:
        raw = f.read()
    if path.endswith(".ipynb"):
        self.content = json.loads(raw)
    else:
        self.content = raw
    return await self.update(context=context, **context_kwargs)

update_from_string async

update_from_string(
    s: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Set content from a string (JSON for .ipynb, plain text otherwise) and call update.

Source code in src/crew_dcs/classes/DomoJupyter/Content.py

async def update_from_string(
    self,
    s: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Set content from a string (JSON for .ipynb, plain text otherwise) and call update."""
    if self.name.endswith(".ipynb"):
        self.content = json.loads(s)
    else:
        self.content = s
    return await self.update(context=context, **context_kwargs)

JupyterNotebook_DataflowError

JupyterNotebook_DataflowError(
    cls: Any = None,
    cls_instance: Any = None,
    entity_id_col: str | None = "id",
    **kwargs
)

Bases: ClassError

Raised when dataflow methods are called without required notebook_id or dj_workspace.

Source code in src/crew_dcs/base/exceptions.py

def __init__(
    self,
    cls: Any = None,
    cls_instance: Any = None,
    entity_id_col: str | None = "id",
    **kwargs,
):
    self.cls = cls
    self.cls_instance = cls_instance
    self.entity_id_col = entity_id_col

    # Extract class-specific information
    if not kwargs.get("entity_id"):
        kwargs["entity_id"] = self._get_entity_id_str()
    if not kwargs.get("parent_class"):
        kwargs["parent_class"] = self._get_parent_class_str()

    super().__init__(**kwargs)