jupyter

jupyter

Jupyter Route Functions

CRUD operations for Domo Jupyter workspaces and their content.

The module is organized into functional areas:

• core — Workspace retrieval, start, and auth-token helpers (domo.com API, standard DomoAuth).
• config — Workspace configuration updates and instance settings (domo.com API, standard DomoAuth).
• content — File / notebook / directory CRUD and checkpoints (Jupyter server API, requires DomoJupyterAuth with jupyter_token).
• exceptions — All Jupyter-related exception classes.

JupyterCheckpoint

Bases: TypedDict

Shape returned by checkpoint endpoints.

Example::

{"id": "checkpoint", "last_modified": "2026-03-16T20:23:31.937000Z"}

JupyterContentItem

Bases: TypedDict

Shape returned by GET /api/contents/{path}.

Example::

{
    "name": "my_notebook.ipynb",
    "path": "folder/my_notebook.ipynb",
    "last_modified": "2026-03-16T20:23:31.937000Z",
    "created": "2026-03-16T20:23:31.937000Z",
    "content": null,          # None when ?content=0
    "format": "json",         # "json" | "text" | null
    "mimetype": null,
    "size": 624,
    "writable": true,
    "hash": null,
    "hash_algorithm": null,
    "type": "notebook"        # "notebook" | "file" | "directory"
}

JupyterCreateBody

Bases: TypedDict

Body for POST /api/contents/{parent_path} to create new content.

The Jupyter API creates content with a default name (e.g. Untitled.ipynb or Untitled Folder). Use rename_jupyter_content afterward to give it the desired name.

Example::

{"type": "notebook", "path": "my_folder"}

JupyterRenameBody

Bases: TypedDict

Body for PATCH /api/contents/{old_path} to rename or move content.

Example::

{"path": "folder/new_name.ipynb"}

JupyterUpdateBody

Bases: TypedDict

Body for PUT /api/contents/{path} to save content.

Example (notebook)::

{
    "type": "notebook",
    "format": "json",
    "content": {"cells": [...], "metadata": {...}, "nbformat": 4, "nbformat_minor": 5}
}

Example (text file)::

{"type": "file", "format": "text", "content": "hello world"}

JupyterWorkspace_Error

JupyterWorkspace_Error(
    operation: str,
    workspace_id: str,
    message: str | None = None,
    res=None,
    **kwargs
)

Bases: RouteError

Raised when Jupyter workspace operations fail.

Source code in src/crew_dcs/routes/jupyter/exceptions.py

def __init__(
    self,
    operation: str,
    workspace_id: str,
    message: str | None = None,
    res=None,
    **kwargs,
):
    super().__init__(
        message=message or f"Jupyter workspace {operation} operation failed",
        entity_id=workspace_id,
        res=res,
        **kwargs,
    )

Jupyter_CRUD_Error

Jupyter_CRUD_Error(
    operation: str,
    workspace_id: str | None = None,
    content_path: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs
)

Bases: RouteError

Raised when Jupyter workspace create, update, or delete operations fail.

Source code in src/crew_dcs/routes/jupyter/exceptions.py

def __init__(
    self,
    operation: str,
    workspace_id: str | None = None,
    content_path: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs,
):
    entity_id = workspace_id or content_path
    super().__init__(
        message=message or f"Jupyter {operation} operation failed",
        entity_id=entity_id,
        res=res,
        **kwargs,
    )

Jupyter_GET_Error

Jupyter_GET_Error(
    workspace_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs
)

Bases: RouteError

Raised when Jupyter workspace retrieval operations fail.

Source code in src/crew_dcs/routes/jupyter/exceptions.py

def __init__(
    self,
    workspace_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs,
):
    super().__init__(
        message=message,
        entity_id=workspace_id,
        res=res,
        **kwargs,
    )

SearchJupyterNotFoundError

SearchJupyterNotFoundError(
    search_criteria: str,
    message: str | None = None,
    res=None,
    **kwargs
)

Bases: RouteError

Raised when Jupyter workspace search operations return no results.

Source code in src/crew_dcs/routes/jupyter/exceptions.py

def __init__(
    self,
    search_criteria: str,
    message: str | None = None,
    res=None,
    **kwargs,
):
    super().__init__(
        message=message
        or f"Jupyter search returned no results for: {search_criteria}",
        entity_id=search_criteria,
        res=res,
        **kwargs,
    )

create_checkpoint async

create_checkpoint(
    auth: DomoJupyterAuth,
    content_path: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Create a checkpoint (snapshot) for a content item.

POST /api/contents/{path}/checkpoints

Parameters:

Name	Type	Description	Default
auth	DomoJupyterAuth	Jupyter authentication object.	required
content_path	str	Path of the content to checkpoint.	required
return_raw	bool	Return raw response without error checking.	False
context	RouteContext | None	Optional RouteContext.	None

Returns:

Type	Description
ResponseGetData	ResponseGetData whose .response is a JupyterCheckpoint.

Raises:

Type	Description
Jupyter_CRUD_Error	On failure.

Source code in src/crew_dcs/routes/jupyter/content.py

@gd.route_function
@log_call(**_LOG_DECORATORS)
async def create_checkpoint(
    auth: dmda.DomoJupyterAuth,
    content_path: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create a checkpoint (snapshot) for a content item.

    ``POST /api/contents/{path}/checkpoints``

    Args:
        auth: Jupyter authentication object.
        content_path: Path of the content to checkpoint.
        return_raw: Return raw response without error checking.
        context: Optional ``RouteContext``.

    Returns:
        ``ResponseGetData`` whose ``.response`` is a ``JupyterCheckpoint``.

    Raises:
        Jupyter_CRUD_Error: On failure.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    dmda.test_is_jupyter_auth(auth)

    url = _build_jupyter_contents_url(auth, f"{content_path}/checkpoints")

    res = await gd.get_data(
        url=url,
        method="POST",
        auth=auth,
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success:
        raise Jupyter_CRUD_Error(
            operation="create_checkpoint",
            content_path=content_path,
            message=f"Failed to create checkpoint for {content_path}",
            res=res,
        )

    return res

create_folder async

create_folder(
    auth: DomoJupyterAuth,
    folder_path: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Create a directory in a Jupyter workspace.

Two-step flow matching the JupyterLab UI:

1. POST /api/contents/{parent} with {"type": "directory"} — creates Untitled Folder.
2. PATCH /api/contents/Untitled%20Folder — renames to folder_path.

Parameters:

Name	Type	Description	Default
auth	DomoJupyterAuth	Jupyter authentication object.	required
folder_path	str	Desired folder path (e.g. "my_project/sub_folder").	required
return_raw	bool	Return raw response after POST.	False
context	RouteContext | None	Optional RouteContext.	None

Returns:

Type	Description
ResponseGetData	ResponseGetData whose .response is a JupyterContentItem
ResponseGetData	with type="directory".

Raises:

Type	Description
Jupyter_CRUD_Error	On creation or rename failure.

Source code in src/crew_dcs/routes/jupyter/content.py

@gd.route_function
@log_call(**_LOG_DECORATORS)
async def create_folder(
    auth: dmda.DomoJupyterAuth,
    folder_path: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create a directory in a Jupyter workspace.

    Two-step flow matching the JupyterLab UI:

    1. ``POST /api/contents/{parent}`` with ``{"type": "directory"}`` — creates
       ``Untitled Folder``.
    2. ``PATCH /api/contents/Untitled%20Folder`` — renames to ``folder_path``.

    Args:
        auth: Jupyter authentication object.
        folder_path: Desired folder path (e.g. ``"my_project/sub_folder"``).
        return_raw: Return raw response after POST.
        context: Optional ``RouteContext``.

    Returns:
        ``ResponseGetData`` whose ``.response`` is a ``JupyterContentItem``
        with ``type="directory"``.

    Raises:
        Jupyter_CRUD_Error: On creation or rename failure.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    dmda.test_is_jupyter_auth(auth)

    if folder_path.startswith("./"):
        folder_path = folder_path[2:]

    parts = os.path.normpath(folder_path).split(os.sep)
    parent_path = "/".join(parts[:-1]) if len(parts) > 1 else ""

    post_body: JupyterCreateBody = {"type": "directory", "path": parent_path}
    parent_url = _build_jupyter_contents_url(auth, parent_path)

    res_post = await gd.get_data(
        url=parent_url,
        method="POST",
        auth=auth,
        body=post_body,
        context=context,
    )

    if return_raw:
        return res_post

    if res_post.status == 403:
        raise Jupyter_CRUD_Error(
            operation="create_folder",
            content_path=folder_path,
            message="Unable to query API, valid jupyter_token?",
            res=res_post,
        )

    if not res_post.is_success:
        raise Jupyter_CRUD_Error(
            operation="create_folder", content_path=folder_path, res=res_post
        )

    untitled_path = res_post.response["path"]

    res = await rename_jupyter_content(
        auth=auth,
        old_path=untitled_path,
        new_path=folder_path,
        context=context,
    )

    res.response = {**res_post.response, **res.response}
    return res

create_jupyter_obj async

create_jupyter_obj(
    auth: DomoJupyterAuth,
    new_content: Any = "",
    content_path: str = "",
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Create new content in a Jupyter workspace.

Performs two API calls:

1. POST /api/contents/{parent} with JupyterCreateBody — creates an object with a default name (e.g. Untitled.ipynb).
2. PATCH /api/contents/{untitled_path} with JupyterRenameBody — renames to the desired content_path.

If new_content is provided, a third PUT call saves the content.

Parameters:

Name	Type	Description	Default
auth	DomoJupyterAuth	Jupyter authentication object.	required
new_content	Any	Content to write after creation (optional).	''
content_path	str	Desired full path (e.g. "my_folder/my_notebook.ipynb").	''
return_raw	bool	Return raw response after the POST step.	False
context	RouteContext | None	Optional RouteContext.	None

Returns:

Type	Description
ResponseGetData	ResponseGetData whose .response is a JupyterContentItem
ResponseGetData	reflecting the final state.

Raises:

Type	Description
Jupyter_CRUD_Error	On creation or rename failure.

Source code in src/crew_dcs/routes/jupyter/content.py

@gd.route_function
@log_call(**_LOG_DECORATORS)
async def create_jupyter_obj(
    auth: dmda.DomoJupyterAuth,
    new_content: Any = "",
    content_path: str = "",
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create new content in a Jupyter workspace.

    Performs two API calls:

    1. ``POST /api/contents/{parent}`` with ``JupyterCreateBody`` — creates an
       object with a default name (e.g. ``Untitled.ipynb``).
    2. ``PATCH /api/contents/{untitled_path}`` with ``JupyterRenameBody`` —
       renames to the desired ``content_path``.

    If ``new_content`` is provided, a third ``PUT`` call saves the content.

    Args:
        auth: Jupyter authentication object.
        new_content: Content to write after creation (optional).
        content_path: Desired full path (e.g. ``"my_folder/my_notebook.ipynb"``).
        return_raw: Return raw response after the POST step.
        context: Optional ``RouteContext``.

    Returns:
        ``ResponseGetData`` whose ``.response`` is a ``JupyterContentItem``
        reflecting the final state.

    Raises:
        Jupyter_CRUD_Error: On creation or rename failure.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    dmda.test_is_jupyter_auth(auth)

    if content_path.startswith("./"):
        content_path = content_path[2:]

    content_type = _infer_content_type(content_path)

    # parent directory where the new object will be created
    parts = os.path.normpath(content_path).split(os.sep)
    parent_path = "/".join(parts[:-1]) if len(parts) > 1 else ""

    post_body: JupyterCreateBody = {"type": content_type, "path": parent_path}

    parent_url = _build_jupyter_contents_url(auth, parent_path)

    res_post = await gd.get_data(
        url=parent_url,
        method="POST",
        auth=auth,
        body=post_body,
        context=context,
    )

    if return_raw:
        return res_post

    if res_post.status == 403:
        raise Jupyter_CRUD_Error(
            operation="create",
            content_path=content_path,
            message="Unable to query API, valid jupyter_token?",
            res=res_post,
        )

    if not res_post.is_success:
        raise Jupyter_CRUD_Error(
            operation="create", content_path=content_path, res=res_post
        )

    # Rename from the default name (e.g. "Untitled.ipynb") to desired path
    untitled_path = res_post.response["path"]

    res = await rename_jupyter_content(
        auth=auth,
        old_path=untitled_path,
        new_path=content_path,
        context=context,
    )

    res.response = {**res_post.response, **res.response}

    # Optionally save content
    if new_content:
        await update_jupyter_file(
            auth=auth,
            new_content=new_content,
            content_path=content_path,
            context=context,
        )

    return res

delete_jupyter_content async

delete_jupyter_content(
    auth: DomoJupyterAuth,
    content_path: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Delete content from a Jupyter workspace.

DELETE /api/contents/{path}

Parameters:

Name	Type	Description	Default
auth	DomoJupyterAuth	Jupyter authentication object.	required
content_path	str	Path to the content to delete.	required
return_raw	bool	Return raw response without error checking.	False
context	RouteContext | None	Optional RouteContext.	None

Returns:

Type	Description
ResponseGetData	ResponseGetData (status 204 on success).

Raises:

Type	Description
Jupyter_CRUD_Error	On 403 or other failure.
SearchJupyterNotFoundError	On 404.

Source code in src/crew_dcs/routes/jupyter/content.py

@gd.route_function
@log_call(**_LOG_DECORATORS)
async def delete_jupyter_content(
    auth: dmda.DomoJupyterAuth,
    content_path: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Delete content from a Jupyter workspace.

    ``DELETE /api/contents/{path}``

    Args:
        auth: Jupyter authentication object.
        content_path: Path to the content to delete.
        return_raw: Return raw response without error checking.
        context: Optional ``RouteContext``.

    Returns:
        ``ResponseGetData`` (status 204 on success).

    Raises:
        Jupyter_CRUD_Error: On 403 or other failure.
        SearchJupyterNotFoundError: On 404.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    dmda.test_is_jupyter_auth(auth)

    url = _build_jupyter_contents_url(auth, content_path)

    res = await gd.get_data(
        url=url,
        method="DELETE",
        auth=auth,
        context=context,
    )

    if return_raw:
        return res

    if res.status == 403:
        raise Jupyter_CRUD_Error(
            operation="delete",
            content_path=content_path,
            message="Unable to query API, valid jupyter_token?",
            res=res,
        )

    if res.status == 404:
        raise SearchJupyterNotFoundError(
            search_criteria=f"content_path: {content_path}", res=res
        )

    if not res.is_success:
        raise Jupyter_CRUD_Error(operation="delete", content_path=content_path, res=res)

    return res

generate_update_jupyter_body

generate_update_jupyter_body(
    new_content: Any, content_path: str
) -> JupyterUpdateBody

Build a JupyterUpdateBody for saving content via PUT.

Determines type and format from content_path.

Parameters:

Name	Type	Description	Default
new_content	Any	The content payload (notebook JSON dict or text string).	required
content_path	str	Path of the content — extension determines type/format.	required

Returns:

Type	Description
JupyterUpdateBody	A dict conforming to JupyterUpdateBody.

Source code in src/crew_dcs/routes/jupyter/content.py

def generate_update_jupyter_body(
    new_content: Any,
    content_path: str,
) -> JupyterUpdateBody:
    """Build a ``JupyterUpdateBody`` for saving content via PUT.

    Determines ``type`` and ``format`` from ``content_path``.

    Args:
        new_content: The content payload (notebook JSON dict or text string).
        content_path: Path of the content — extension determines type/format.

    Returns:
        A dict conforming to ``JupyterUpdateBody``.
    """
    if content_path.startswith("./"):
        content_path = content_path[2:]

    content_type = _infer_content_type(content_path)

    return {
        "type": content_type,
        "format": _infer_format(content_type),
        "content": new_content,
    }

get_checkpoints async

get_checkpoints(
    auth: DomoJupyterAuth,
    content_path: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

List checkpoints for a content item.

GET /api/contents/{path}/checkpoints

Parameters:

Name	Type	Description	Default
auth	DomoJupyterAuth	Jupyter authentication object.	required
content_path	str	Path of the content.	required
return_raw	bool	Return raw response without error checking.	False
context	RouteContext | None	Optional RouteContext.	None

Returns:

Type	Description
ResponseGetData	ResponseGetData whose .response is list[JupyterCheckpoint].

Raises:

Type	Description
Jupyter_GET_Error	On failure.

Source code in src/crew_dcs/routes/jupyter/content.py

@gd.route_function
@log_call(**_LOG_DECORATORS)
async def get_checkpoints(
    auth: dmda.DomoJupyterAuth,
    content_path: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """List checkpoints for a content item.

    ``GET /api/contents/{path}/checkpoints``

    Args:
        auth: Jupyter authentication object.
        content_path: Path of the content.
        return_raw: Return raw response without error checking.
        context: Optional ``RouteContext``.

    Returns:
        ``ResponseGetData`` whose ``.response`` is ``list[JupyterCheckpoint]``.

    Raises:
        Jupyter_GET_Error: On failure.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    dmda.test_is_jupyter_auth(auth)

    url = _build_jupyter_contents_url(auth, f"{content_path}/checkpoints")

    res = await gd.get_data(
        url=url,
        method="GET",
        auth=auth,
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success:
        raise Jupyter_GET_Error(
            message=f"Failed to get checkpoints for {content_path}", res=res
        )

    return res

get_content async

get_content(
    auth: DomoJupyterAuth,
    content_path: str = "",
    ignore_folders: list[str] | None = None,
    included_filetypes: list[str] | None = None,
    is_recursive: bool = True,
    exclude_hidden: bool = True,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Get all content from a Jupyter workspace, optionally recursive.

Convenience wrapper around get_content_recursive with automatic deduplication.

Parameters:

Name	Type	Description	Default
auth	DomoJupyterAuth	Jupyter authentication object.	required
content_path	str	Starting path ("" = workspace root).	''
ignore_folders	list[str] | None	Folder names to exclude (matches path segments).	None
included_filetypes	list[str] | None	File extensions to include (e.g. ['.ipynb', '.py']).	None
is_recursive	bool	Descend into subdirectories (default True).	True
exclude_hidden	bool	Skip dot-prefixed file and directory names (default True).	True
return_raw	bool	Pass through to underlying calls.	False
context	RouteContext | None	Optional RouteContext.	None

Returns:

Type	Description
ResponseGetData	ResponseGetData whose .response is a flat
ResponseGetData	list[JupyterContentItem].

Raises:

Type	Description
Jupyter_GET_Error	On retrieval failure.
SearchJupyterNotFoundError	If starting path does not exist.

Source code in src/crew_dcs/routes/jupyter/content.py

@gd.route_function
@log_call(**_LOG_DECORATORS)
async def get_content(
    auth: dmda.DomoJupyterAuth,
    content_path: str = "",
    ignore_folders: list[str] | None = None,
    included_filetypes: list[str] | None = None,
    is_recursive: bool = True,
    exclude_hidden: bool = True,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get all content from a Jupyter workspace, optionally recursive.

    Convenience wrapper around ``get_content_recursive`` with automatic
    deduplication.

    Args:
        auth: Jupyter authentication object.
        content_path: Starting path (``""`` = workspace root).
        ignore_folders: Folder names to exclude (matches path segments).
        included_filetypes: File extensions to include (e.g. ``['.ipynb', '.py']``).
        is_recursive: Descend into subdirectories (default ``True``).
        exclude_hidden: Skip dot-prefixed file and directory names (default ``True``).
        return_raw: Pass through to underlying calls.
        context: Optional ``RouteContext``.

    Returns:
        ``ResponseGetData`` whose ``.response`` is a flat
        ``list[JupyterContentItem]``.

    Raises:
        Jupyter_GET_Error: On retrieval failure.
        SearchJupyterNotFoundError: If starting path does not exist.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    dmda.test_is_jupyter_auth(auth)

    all_rows: list = []
    seen_paths: set = set()
    res = None

    return await get_content_recursive(
        auth=auth,
        content_path=content_path,
        all_rows=all_rows,
        res=res,
        seen_paths=seen_paths,
        ignore_folders=ignore_folders,
        included_filetypes=included_filetypes,
        is_recursive=is_recursive,
        exclude_hidden=exclude_hidden,
        is_run_test_jupyter_auth=False,
        return_raw=return_raw,
        context=context,
    )

get_content_recursive async

get_content_recursive(
    auth: DomoJupyterAuth,
    all_rows: list,
    content_path: str,
    res: ResponseGetData,
    seen_paths: set,
    obj: dict | None = None,
    ignore_folders: list[str] | None = None,
    included_filetypes: list[str] | None = None,
    is_recursive: bool = True,
    exclude_hidden: bool = True,
    is_run_test_jupyter_auth: bool = True,
    return_raw: bool = False,
    context: RouteContext | None = None,
    **context_kwargs
)

Recursively retrieve content from a Jupyter workspace.

This is an internal accumulator function called by get_content.

Parameters:

Name	Type	Description	Default
auth	DomoJupyterAuth	Jupyter authentication object.	required
all_rows	list	Accumulator list for all content items found so far.	required
content_path	str	Current path being processed.	required
res	ResponseGetData	ResponseGetData object to update with accumulated results.	required
seen_paths	set	Set of already-visited paths (deduplication).	required
obj	dict | None	Pre-fetched content dict (None on first call).	None
ignore_folders	list[str] | None	Folder names to skip (matched against path segments).	None
included_filetypes	list[str] | None	File extensions to include (e.g. ['.ipynb']).	None
is_recursive	bool	Whether to descend into subdirectories.	True
exclude_hidden	bool	If True, skip files and directories whose Jupyter name starts with . (e.g. .git, .env).	True
is_run_test_jupyter_auth	bool	Validate auth on the first call.	True
return_raw	bool	Pass through to get_jupyter_content.	False
context	RouteContext | None	Optional RouteContext.	None

Returns:

Type	Description
	ResponseGetData with .response set to all_rows.

Source code in src/crew_dcs/routes/jupyter/content.py

async def get_content_recursive(  # noqa: C901
    auth: dmda.DomoJupyterAuth,
    all_rows: list,
    content_path: str,
    res: rgd.ResponseGetData,
    seen_paths: set,
    obj: dict | None = None,
    ignore_folders: list[str] | None = None,
    included_filetypes: list[str] | None = None,
    is_recursive: bool = True,
    exclude_hidden: bool = True,
    is_run_test_jupyter_auth: bool = True,
    return_raw: bool = False,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Recursively retrieve content from a Jupyter workspace.

    This is an internal accumulator function called by ``get_content``.

    Args:
        auth: Jupyter authentication object.
        all_rows: Accumulator list for all content items found so far.
        content_path: Current path being processed.
        res: ``ResponseGetData`` object to update with accumulated results.
        seen_paths: Set of already-visited paths (deduplication).
        obj: Pre-fetched content dict (``None`` on first call).
        ignore_folders: Folder names to skip (matched against path segments).
        included_filetypes: File extensions to include (e.g. ``['.ipynb']``).
        is_recursive: Whether to descend into subdirectories.
        exclude_hidden: If True, skip files and directories whose Jupyter ``name``
            starts with ``.`` (e.g. ``.git``, ``.env``).
        is_run_test_jupyter_auth: Validate auth on the first call.
        return_raw: Pass through to ``get_jupyter_content``.
        context: Optional ``RouteContext``.

    Returns:
        ``ResponseGetData`` with ``.response`` set to ``all_rows``.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    ignore_folders = ignore_folders or []
    included_filetypes = included_filetypes or []

    if not obj:
        obj_res = await get_jupyter_content(
            auth=auth,
            content_path=content_path,
            return_raw=return_raw,
            is_run_test_jupyter_auth=is_run_test_jupyter_auth,
            context=context,
        )
        obj = obj_res.response
        if not res:
            res = obj_res

    obj_path = obj.get("path", "")
    if obj_path in seen_paths:
        return res

    seen_paths.add(obj_path)
    all_rows.append(obj)

    if obj.get("type") != "directory" or not is_recursive:
        res.response = all_rows
        return res

    obj_content = obj.get("content", [])

    filtered_content = []
    for item in obj_content:
        if not isinstance(item, dict):
            continue

        item_name = item.get("name", "")
        item_path = item.get("path", "")
        item_type = item.get("type", "")

        if item_path in seen_paths:
            continue

        if item_name == ".ipynb_checkpoints":
            continue

        if exclude_hidden and _jupyter_item_name_is_hidden(item_name):
            continue

        if ignore_folders and any(
            ign in item_path.split("/") for ign in ignore_folders
        ):
            continue

        if "recent_executions" in item_path:
            continue

        if item_type == "directory":
            filtered_content.append(item)
            continue

        if included_filetypes:
            if any(item_name.endswith(ext) for ext in included_filetypes):
                filtered_content.append(item)
        else:
            filtered_content.append(item)

    res.response = all_rows

    if filtered_content:
        base_drop = context.debug_num_stacks_to_drop if context else 1
        recursive_context = RouteContext.build_context(
            context=context,
            debug_num_stacks_to_drop=base_drop + 1,
            **context_kwargs,
        )
        await dmce.gather_with_concurrency(
            *[
                get_content_recursive(
                    auth=auth,
                    content_path=item["path"],
                    all_rows=all_rows,
                    res=res,
                    seen_paths=seen_paths,
                    ignore_folders=ignore_folders,
                    included_filetypes=included_filetypes,
                    is_recursive=is_recursive,
                    exclude_hidden=exclude_hidden,
                    is_run_test_jupyter_auth=False,
                    return_raw=return_raw,
                    context=recursive_context,
                )
                for item in filtered_content
            ],
            n=5,
        )

    return res

get_jupyter_content async

get_jupyter_content(
    auth: DomoJupyterAuth,
    content_path: str = "",
    include_content: bool = True,
    is_run_test_jupyter_auth: bool = True,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve content metadata (and optionally body) from a Jupyter workspace.

Parameters:

Name	Type	Description	Default
auth	DomoJupyterAuth	Jupyter authentication object.	required
content_path	str	Path within the workspace ("" = root).	''
include_content	bool	If True (default), response includes full content body / directory children (?content=1). Set to False for metadata only (?content=0).	True
is_run_test_jupyter_auth	bool	Validate auth is DomoJupyterAuth.	True
return_raw	bool	Return the raw ResponseGetData without error checking.	False
context	RouteContext | None	Optional RouteContext.	None

Returns:

Type	Description
ResponseGetData	ResponseGetData whose .response is a JupyterContentItem.

Raises:

Type	Description
Jupyter_GET_Error	On 403 or other non-success status.
SearchJupyterNotFoundError	On 404.

Source code in src/crew_dcs/routes/jupyter/content.py

@gd.route_function
@log_call(**_LOG_DECORATORS)
async def get_jupyter_content(
    auth: dmda.DomoJupyterAuth,
    content_path: str = "",
    include_content: bool = True,
    is_run_test_jupyter_auth: bool = True,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve content metadata (and optionally body) from a Jupyter workspace.

    Args:
        auth: Jupyter authentication object.
        content_path: Path within the workspace (``""`` = root).
        include_content: If ``True`` (default), response includes full content
            body / directory children (``?content=1``).  Set to ``False`` for
            metadata only (``?content=0``).
        is_run_test_jupyter_auth: Validate auth is ``DomoJupyterAuth``.
        return_raw: Return the raw ``ResponseGetData`` without error checking.
        context: Optional ``RouteContext``.

    Returns:
        ``ResponseGetData`` whose ``.response`` is a ``JupyterContentItem``.

    Raises:
        Jupyter_GET_Error: On 403 or other non-success status.
        SearchJupyterNotFoundError: On 404.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    if is_run_test_jupyter_auth:
        dmda.test_is_jupyter_auth(auth)

    url = _build_jupyter_contents_url(auth, content_path)

    res = await gd.get_data(
        url=url,
        method="GET",
        auth=auth,
        params={"content": "1" if include_content else "0"},
        context=context,
    )

    if return_raw:
        return res

    if res.status == 403:
        raise Jupyter_GET_Error(
            message="Unable to query API, valid jupyter_token?", res=res
        )

    if res.status == 404:
        raise SearchJupyterNotFoundError(
            search_criteria=f"content_path: {content_path}", res=res
        )

    if not res.is_success:
        raise Jupyter_GET_Error(message="Failed to retrieve Jupyter content", res=res)

    return res

get_jupyter_settings async

get_jupyter_settings(
    auth: DomoAuth,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve Jupyter settings for the Domo instance.

GET /api/datascience/v1/settings

Returns instance types, limits, kernel specs, feature flags, etc.

Example response shape::

{
    "instanceTypes": [{"cpu": 0.5, "memory": 4.0}, ...],
    "limits": {...},
    "isEnabled": true,
    "userAccess": {...},
    "currentUserInstanceTypes": [...],
    "jupyterKernels": [...],
    "featureFlags": {...},
    "locale": "en-US",
    "timeZone": "America/Denver",
    "maintenance": {...}
}

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object (standard DomoAuth).	required
return_raw	bool	Return raw response without error checking.	False
context	RouteContext | None	Optional RouteContext.	None

Returns:

Type	Description
ResponseGetData	ResponseGetData with settings dict.

Raises:

Type	Description
Jupyter_GET_Error	On failure.

Source code in src/crew_dcs/routes/jupyter/config.py

@gd.route_function
@log_call(**_LOG_DECORATORS)
async def get_jupyter_settings(
    auth: DomoAuth,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve Jupyter settings for the Domo instance.

    ``GET /api/datascience/v1/settings``

    Returns instance types, limits, kernel specs, feature flags, etc.

    Example response shape::

        {
            "instanceTypes": [{"cpu": 0.5, "memory": 4.0}, ...],
            "limits": {...},
            "isEnabled": true,
            "userAccess": {...},
            "currentUserInstanceTypes": [...],
            "jupyterKernels": [...],
            "featureFlags": {...},
            "locale": "en-US",
            "timeZone": "America/Denver",
            "maintenance": {...}
        }

    Args:
        auth: Authentication object (standard ``DomoAuth``).
        return_raw: Return raw response without error checking.
        context: Optional ``RouteContext``.

    Returns:
        ``ResponseGetData`` with settings dict.

    Raises:
        Jupyter_GET_Error: On failure.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/datascience/v1/settings"

    res = await gd.get_data(
        url=url,
        method="GET",
        auth=auth,
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success:
        raise Jupyter_GET_Error(message="Failed to retrieve Jupyter settings", res=res)

    return res

get_jupyter_user_namespace async

get_jupyter_user_namespace(
    auth: DomoAuth,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve the Jupyter user namespace for the current user.

GET /api/datascience/v1/workspaces/user

The namespace string (e.g. "my-company-config-672596669") appears in the service_prefix path of all Jupyter server URLs: /user/{namespace}/{workspace_short_id}/api/contents/...

Example response::

{"name": "my-company-config-672596669"}

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object (standard DomoAuth).	required
return_raw	bool	Return raw response without error checking.	False
context	RouteContext | None	Optional RouteContext.	None

Returns:

Type	Description
ResponseGetData	ResponseGetData whose .response is {"name": str}.

Raises:

Type	Description
Jupyter_GET_Error	On failure.

Source code in src/crew_dcs/routes/jupyter/config.py

@gd.route_function
@log_call(**_LOG_DECORATORS)
async def get_jupyter_user_namespace(
    auth: DomoAuth,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve the Jupyter user namespace for the current user.

    ``GET /api/datascience/v1/workspaces/user``

    The namespace string (e.g. ``"my-company-config-672596669"``) appears in
    the ``service_prefix`` path of all Jupyter server URLs:
    ``/user/{namespace}/{workspace_short_id}/api/contents/...``

    Example response::

        {"name": "my-company-config-672596669"}

    Args:
        auth: Authentication object (standard ``DomoAuth``).
        return_raw: Return raw response without error checking.
        context: Optional ``RouteContext``.

    Returns:
        ``ResponseGetData`` whose ``.response`` is ``{"name": str}``.

    Raises:
        Jupyter_GET_Error: On failure.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/datascience/v1/workspaces/user"

    res = await gd.get_data(
        url=url,
        method="GET",
        auth=auth,
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success:
        raise Jupyter_GET_Error(
            message="Failed to retrieve Jupyter user namespace", res=res
        )

    return res

get_jupyter_workspace_by_id async

get_jupyter_workspace_by_id(
    auth: DomoAuth,
    workspace_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve a specific Jupyter workspace by ID.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
workspace_id	str	Unique identifier for the workspace to retrieve	required
return_raw	bool	Return raw API response without processing	False
context	RouteContext | None	RouteContext for request configuration	None
**context_kwargs		Additional context parameters (session, debug_api, etc.)	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing workspace details

Raises:

Type	Description
Jupyter_GET_Error	If workspace retrieval fails

Source code in src/crew_dcs/routes/jupyter/core.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_jupyter_workspace_by_id(
    auth: DomoAuth,
    workspace_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve a specific Jupyter workspace by ID.

    Args:
        auth: Authentication object containing credentials and instance info
        workspace_id: Unique identifier for the workspace to retrieve
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing workspace details

    Raises:
        Jupyter_GET_Error: If workspace retrieval fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/datascience/v1/workspaces/{workspace_id}"

    res = await gd.get_data(
        url=url,
        method="GET",
        auth=auth,
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success:
        raise Jupyter_GET_Error(
            workspace_id=workspace_id,
            message=f"Failed to retrieve workspace {workspace_id}",
            res=res,
        )

    return res

get_jupyter_workspaces async

get_jupyter_workspaces(
    auth: DomoAuth,
    return_raw: bool = False,
    debug_loop: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve all available Jupyter workspaces.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
context	RouteContext | None	Optional RouteContext for request configuration	None
return_raw	bool	Return raw API response without processing	False
debug_loop	bool	Enable detailed loop debugging	False
**context_kwargs		Additional context parameters (session, debug_api, etc.)	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing list of Jupyter workspaces

Raises:

Type	Description
Jupyter_GET_Error	If workspace retrieval fails

Source code in src/crew_dcs/routes/jupyter/core.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_jupyter_workspaces(
    auth: DomoAuth,
    return_raw: bool = False,
    debug_loop: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve all available Jupyter workspaces.

    Args:
        auth: Authentication object containing credentials and instance info
        context: Optional RouteContext for request configuration
        return_raw: Return raw API response without processing
        debug_loop: Enable detailed loop debugging
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing list of Jupyter workspaces

    Raises:
        Jupyter_GET_Error: If workspace retrieval fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/datascience/v1/search/workspaces"

    body = {
        "limit": 50,
        "offset": 0,
        "sortFieldMap": {"CREATED": "DESC"},
        "filters": [],
    }

    def arr_fn(res):
        return res.response["workspaces"]

    offset_params = {"limit": "limit", "offset": "offset"}

    res = await gd.looper(
        url=url,
        method="POST",
        limit=50,
        body=body,
        auth=auth,
        arr_fn=arr_fn,
        offset_params_in_body=True,
        offset_params=offset_params,
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success:
        raise Jupyter_GET_Error(
            message="Failed to retrieve Jupyter workspaces", res=res
        )

    return res

get_notebook_dataflow async

get_notebook_dataflow(
    auth: DomoJupyterAuth,
    workspace_id: str,
    notebook_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Get dataflow (scheduling) configuration for a notebook in a Jupyter workspace.

GET .../services/domo/v1/workspaces/{workspace_id}/notebooks/{notebook_id}/dataflow

Parameters:

Name	Type	Description	Default
auth	DomoJupyterAuth	Jupyter authentication object.	required
workspace_id	str	Workspace UUID (e.g. from get_jupyter_workspace_by_id).	required
notebook_id	str	Notebook UUID (e.g. from content metadata if exposed by API).	required
return_raw	bool	Return raw response without error checking.	False
context	RouteContext | None	Optional RouteContext.	None

Returns:

Type	Description
ResponseGetData	ResponseGetData whose .response is the dataflow/scheduling payload.

Raises:

Type	Description
Jupyter_GET_Error	On failure.
SearchJupyterNotFoundError	On 404.

Source code in src/crew_dcs/routes/jupyter/content.py

@gd.route_function
@log_call(**_LOG_DECORATORS)
async def get_notebook_dataflow(
    auth: dmda.DomoJupyterAuth,
    workspace_id: str,
    notebook_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get dataflow (scheduling) configuration for a notebook in a Jupyter workspace.

    ``GET .../services/domo/v1/workspaces/{workspace_id}/notebooks/{notebook_id}/dataflow``

    Args:
        auth: Jupyter authentication object.
        workspace_id: Workspace UUID (e.g. from get_jupyter_workspace_by_id).
        notebook_id: Notebook UUID (e.g. from content metadata if exposed by API).
        return_raw: Return raw response without error checking.
        context: Optional ``RouteContext``.

    Returns:
        ``ResponseGetData`` whose ``.response`` is the dataflow/scheduling payload.

    Raises:
        Jupyter_GET_Error: On failure.
        SearchJupyterNotFoundError: On 404.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    dmda.test_is_jupyter_auth(auth)

    url = _build_jupyter_notebook_dataflow_url(auth, workspace_id, notebook_id)

    res = await gd.get_data(
        url=url,
        method="GET",
        auth=auth,
        context=context,
    )

    if return_raw:
        return res

    if res.status == 404:
        raise SearchJupyterNotFoundError(
            search_criteria=f"workspace_id: {workspace_id}, notebook_id: {notebook_id}",
            res=res,
        )

    if not res.is_success:
        raise Jupyter_GET_Error(
            message=f"Failed to get notebook dataflow for {notebook_id}", res=res
        )

    return res

get_workspace_auth_token_params async

get_workspace_auth_token_params(
    workspace_id: str,
    auth: DomoAuth,
    return_raw: bool = False,
) -> ResponseGetData

params are needed for authenticating requests inside the workspace environment Note: you'll also need a internally generated jupyter_token to authenticate requests returns { service_location , service_prefix}

Source code in src/crew_dcs/routes/jupyter/core.py

async def get_workspace_auth_token_params(
    workspace_id: str, auth: DomoAuth, return_raw: bool = False
) -> rgd.ResponseGetData:
    """
    params are needed for authenticating requests inside the workspace environment
    Note: you'll also need a internally generated jupyter_token to authenticate requests
    returns { service_location , service_prefix}
    """
    res = await get_jupyter_workspace_by_id(workspace_id=workspace_id, auth=auth)

    open_instances = res.response.get("instances")

    if return_raw:
        return open_instances

    if not open_instances:
        raise JupyterWorkspace_Error(
            operation="get_auth_token",
            workspace_id=workspace_id,
            message="There are no open instances. Do you need to start the workspace?",
            res=res,
        )

    return parse_instance_service_location_and_prefix(
        open_instances[0], auth.domo_instance
    )

parse_instance_service_location_and_prefix

parse_instance_service_location_and_prefix(
    instance_dict: dict, domo_instance: str
) -> dict

Parse service location and prefix from instance dictionary.

Source code in src/crew_dcs/routes/jupyter/core.py

def parse_instance_service_location_and_prefix(
    instance_dict: dict, domo_instance: str
) -> dict:
    """Parse service location and prefix from instance dictionary."""
    url = instance_dict["url"]

    query = urllib.parse.unquote(urllib.parse.urlparse(url).query)
    query = urllib.parse.urlparse(query.split("&")[1].replace("next=", ""))

    return {
        "service_location": query.netloc.replace(domo_instance, "")[1:],
        "service_prefix": query.path,
    }

rename_jupyter_content async

rename_jupyter_content(
    auth: DomoJupyterAuth,
    old_path: str,
    new_path: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Rename or move content within a Jupyter workspace.

PATCH /api/contents/{old_path}

Parameters:

Name	Type	Description	Default
auth	DomoJupyterAuth	Jupyter authentication object.	required
old_path	str	Current path of the content.	required
new_path	str	Desired new path.	required
return_raw	bool	Return raw response without error checking.	False
context	RouteContext | None	Optional RouteContext.	None

Returns:

Type	Description
ResponseGetData	ResponseGetData whose .response is a JupyterContentItem
ResponseGetData	with the updated path.

Raises:

Type	Description
Jupyter_CRUD_Error	On 403, 409 (conflict), or other failure.

Source code in src/crew_dcs/routes/jupyter/content.py

@gd.route_function
@log_call(**_LOG_DECORATORS)
async def rename_jupyter_content(
    auth: dmda.DomoJupyterAuth,
    old_path: str,
    new_path: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Rename or move content within a Jupyter workspace.

    ``PATCH /api/contents/{old_path}``

    Args:
        auth: Jupyter authentication object.
        old_path: Current path of the content.
        new_path: Desired new path.
        return_raw: Return raw response without error checking.
        context: Optional ``RouteContext``.

    Returns:
        ``ResponseGetData`` whose ``.response`` is a ``JupyterContentItem``
        with the updated path.

    Raises:
        Jupyter_CRUD_Error: On 403, 409 (conflict), or other failure.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    dmda.test_is_jupyter_auth(auth)

    url = _build_jupyter_contents_url(auth, old_path)
    body: JupyterRenameBody = {"path": new_path}

    res = await gd.get_data(
        url=url,
        method="PATCH",
        auth=auth,
        body=body,
        context=context,
    )

    if return_raw:
        return res

    if res.status == 403:
        raise Jupyter_CRUD_Error(
            operation="rename",
            content_path=old_path,
            message="Unable to query API, valid jupyter_token?",
            res=res,
        )

    if res.status == 409:
        raise Jupyter_CRUD_Error(
            operation="rename",
            content_path=old_path,
            message=f"Conflict — target path '{new_path}' may already exist.",
            res=res,
        )

    if not res.is_success:
        raise Jupyter_CRUD_Error(operation="rename", content_path=old_path, res=res)

    return res

resolve_compute_tier

resolve_compute_tier(
    cpu: float | int | None,
    memory: float | int | None,
    instance_types: list[dict] | None = None,
) -> str | None

Resolve a human-readable compute tier label for a workspace's cpu/memory.

Domo's instanceTypes (returned by both get_jupyter_settings and the per-workspace GET /api/datascience/v1/workspaces/{id} response) only carries {"cpu": ..., "memory": ...} pairs — there is no name/label field to surface as-is (confirmed against a live instance while implementing this — see issue #1497). The label is therefore synthesized as "{cpu} CPU / {memory} GB".

When instance_types is supplied, the cpu/memory pair must match one of the listed tiers or None is returned — this distinguishes a known/standard tier from a custom or unrecognized allocation. When instance_types is omitted, the label is synthesized unconditionally.

Parameters:

Name	Type	Description	Default
cpu	float | int | None	Workspace cpu allocation (e.g. 0.5).	required
memory	float | int | None	Workspace memory allocation in GB (e.g. 4.0).	required
instance_types	list[dict] | None	Optional list of {"cpu": ..., "memory": ...} dicts (from workspace instanceTypes or settings instanceTypes/currentUserInstanceTypes) to validate against.	None

Returns:

Type	Description
str | None	A label like "0.5 CPU / 4 GB", or None if cpu/memory is
str | None	missing, or if instance_types was provided and no entry matches.

Source code in src/crew_dcs/routes/jupyter/config.py

def resolve_compute_tier(
    cpu: float | int | None,
    memory: float | int | None,
    instance_types: list[dict] | None = None,
) -> str | None:
    """Resolve a human-readable compute tier label for a workspace's cpu/memory.

    Domo's ``instanceTypes`` (returned by both ``get_jupyter_settings`` and
    the per-workspace ``GET /api/datascience/v1/workspaces/{id}`` response)
    only carries ``{"cpu": ..., "memory": ...}`` pairs — there is no
    ``name``/``label`` field to surface as-is (confirmed against a live
    instance while implementing this — see issue #1497). The label is
    therefore synthesized as ``"{cpu} CPU / {memory} GB"``.

    When ``instance_types`` is supplied, the cpu/memory pair must match one
    of the listed tiers or ``None`` is returned — this distinguishes a
    known/standard tier from a custom or unrecognized allocation. When
    ``instance_types`` is omitted, the label is synthesized unconditionally.

    Args:
        cpu: Workspace cpu allocation (e.g. ``0.5``).
        memory: Workspace memory allocation in GB (e.g. ``4.0``).
        instance_types: Optional list of ``{"cpu": ..., "memory": ...}``
            dicts (from workspace ``instanceTypes`` or settings
            ``instanceTypes``/``currentUserInstanceTypes``) to validate against.

    Returns:
        A label like ``"0.5 CPU / 4 GB"``, or ``None`` if cpu/memory is
        missing, or if ``instance_types`` was provided and no entry matches.
    """
    if cpu is None or memory is None:
        return None

    if instance_types is not None:
        is_known = any(
            it.get("cpu") == cpu and it.get("memory") == memory for it in instance_types
        )
        if not is_known:
            return None

    return f"{_format_number(cpu)} CPU / {_format_number(memory)} GB"

resolve_python_version

resolve_python_version(
    jupyter_kernel: str | None,
) -> str | None

Parse a Python version string out of a Domo jupyterKernel code.

The workspace API (GET /api/datascience/v1/workspaces/{id}) returns the active kernel as a code such as "PYTHON_3_12" or "R_4_1" on the workspace itself (confirmed present on live workspaces regardless of whether the workspace is currently running — see issue #1497). This parses that code directly rather than requiring a separate get_jupyter_settings call to look the code up in jupyterKernels.

Parameters:

Name	Type	Description	Default
jupyter_kernel	str | None	Raw kernel code, e.g. "PYTHON_3_12".	required

Returns:

Type	Description
str | None	A version string like "3.12" for Python kernels, or None for
str | None	non-Python kernels (e.g. R) or a missing value.

Source code in src/crew_dcs/routes/jupyter/config.py

def resolve_python_version(jupyter_kernel: str | None) -> str | None:
    """Parse a Python version string out of a Domo ``jupyterKernel`` code.

    The workspace API (``GET /api/datascience/v1/workspaces/{id}``) returns
    the active kernel as a code such as ``"PYTHON_3_12"`` or ``"R_4_1"`` on
    the workspace itself (confirmed present on live workspaces regardless of
    whether the workspace is currently running — see issue #1497). This
    parses that code directly rather than requiring a separate
    ``get_jupyter_settings`` call to look the code up in ``jupyterKernels``.

    Args:
        jupyter_kernel: Raw kernel code, e.g. ``"PYTHON_3_12"``.

    Returns:
        A version string like ``"3.12"`` for Python kernels, or ``None`` for
        non-Python kernels (e.g. R) or a missing value.
    """
    if not jupyter_kernel or not jupyter_kernel.upper().startswith("PYTHON_"):
        return None

    version_parts = jupyter_kernel.split("_")[1:]
    if not version_parts:
        return None

    return ".".join(version_parts)

start_jupyter_workspace async

start_jupyter_workspace(
    auth: DomoAuth,
    workspace_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Start a Jupyter workspace instance.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
workspace_id	str	Unique identifier for the workspace to start	required
context	RouteContext | None	Optional RouteContext for request configuration	None
return_raw	bool	Return raw API response without processing	False
**context_kwargs		Additional context parameters (session, debug_api, etc.)	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing workspace start result

Raises:

Type	Description
JupyterWorkspace_Error	If workspace start operation fails
Jupyter_GET_Error	If workspace retrieval fails

Source code in src/crew_dcs/routes/jupyter/core.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def start_jupyter_workspace(
    auth: DomoAuth,
    workspace_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Start a Jupyter workspace instance.

    Args:
        auth: Authentication object containing credentials and instance info
        workspace_id: Unique identifier for the workspace to start
        context: Optional RouteContext for request configuration
        return_raw: Return raw API response without processing
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing workspace start result

    Raises:
        JupyterWorkspace_Error: If workspace start operation fails
        Jupyter_GET_Error: If workspace retrieval fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/datascience/v1/workspaces/{workspace_id}/instances"

    try:
        res = await gd.get_data(
            url=url,
            method="POST",
            auth=auth,
            context=context,
        )

        if return_raw:
            return res

    except RuntimeError as e:
        return rgd.ResponseGetData(
            status=500,
            response=f"starting workspace, please wait - {e}",
            is_success=False,
        )

    if res.status == 500 or res.status == 403:
        raise JupyterWorkspace_Error(
            operation="start",
            workspace_id=workspace_id,
            message=f"You may not have access to this workspace {workspace_id}, is it shared with you? Or may already be started",
            res=res,
        )

    if not res.is_success:
        raise JupyterWorkspace_Error(
            operation="start", workspace_id=workspace_id, res=res
        )

    res.response = "workspace started"
    return res

update_jupyter_file async

update_jupyter_file(
    auth: DomoJupyterAuth,
    new_content: Any,
    content_path: str = "",
    body: dict | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Update (save) content in a Jupyter workspace file.

PUT /api/contents/{path}

Parameters:

Name	Type	Description	Default
auth	DomoJupyterAuth	Jupyter authentication object.	required
new_content	Any	New content payload (notebook JSON dict or text string).	required
content_path	str	Path of the file to update.	''
body	dict | None	Optional fully-formed request body. When provided, new_content is ignored and this dict is sent as-is.	None
return_raw	bool	Return raw response without error checking.	False
context	RouteContext | None	Optional RouteContext.	None

Returns:

Type	Description
ResponseGetData	ResponseGetData whose .response is a JupyterContentItem.

Raises:

Type	Description
Jupyter_CRUD_Error	On 403 or other failure.
SearchJupyterNotFoundError	On 404.

Source code in src/crew_dcs/routes/jupyter/content.py

@gd.route_function
@log_call(**_LOG_DECORATORS)
async def update_jupyter_file(
    auth: dmda.DomoJupyterAuth,
    new_content: Any,
    content_path: str = "",
    body: dict | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Update (save) content in a Jupyter workspace file.

    ``PUT /api/contents/{path}``

    Args:
        auth: Jupyter authentication object.
        new_content: New content payload (notebook JSON dict or text string).
        content_path: Path of the file to update.
        body: Optional fully-formed request body.  When provided,
            ``new_content`` is ignored and this dict is sent as-is.
        return_raw: Return raw response without error checking.
        context: Optional ``RouteContext``.

    Returns:
        ``ResponseGetData`` whose ``.response`` is a ``JupyterContentItem``.

    Raises:
        Jupyter_CRUD_Error: On 403 or other failure.
        SearchJupyterNotFoundError: On 404.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    dmda.test_is_jupyter_auth(auth)

    body = body or generate_update_jupyter_body(new_content, content_path)

    url = _build_jupyter_contents_url(auth, content_path)

    res = await gd.get_data(
        url=url,
        method="PUT",
        auth=auth,
        body=body,
        context=context,
    )

    if return_raw:
        return res

    if res.status == 403:
        raise Jupyter_CRUD_Error(
            operation="update",
            content_path=content_path,
            message="Unable to query API, valid jupyter_token?",
            res=res,
        )

    if res.status == 404:
        raise SearchJupyterNotFoundError(
            search_criteria=f"content_path: {content_path}", res=res
        )

    if not res.is_success:
        raise Jupyter_CRUD_Error(operation="update", content_path=content_path, res=res)

    return res

update_jupyter_workspace_config async

update_jupyter_workspace_config(
    auth: DomoAuth,
    workspace_id: str,
    config: dict,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Update the configuration of a Jupyter workspace.

PUT /api/datascience/v1/workspaces/{workspace_id}

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object.	required
workspace_id	str	UUID of the workspace.	required
config	dict	Configuration dict (inputConfiguration, outputConfiguration, etc.).	required
return_raw	bool	Return raw response without error checking.	False
context	RouteContext | None	Optional RouteContext.	None

Returns:

Type	Description
ResponseGetData	ResponseGetData with updated workspace configuration.

Raises:

Type	Description
Jupyter_CRUD_Error	On update failure.
SearchJupyterNotFoundError	If workspace does not exist (404).

Source code in src/crew_dcs/routes/jupyter/config.py

@gd.route_function
@log_call(**_LOG_DECORATORS)
async def update_jupyter_workspace_config(
    auth: DomoAuth,
    workspace_id: str,
    config: dict,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Update the configuration of a Jupyter workspace.

    ``PUT /api/datascience/v1/workspaces/{workspace_id}``

    Args:
        auth: Authentication object.
        workspace_id: UUID of the workspace.
        config: Configuration dict (inputConfiguration, outputConfiguration, etc.).
        return_raw: Return raw response without error checking.
        context: Optional ``RouteContext``.

    Returns:
        ``ResponseGetData`` with updated workspace configuration.

    Raises:
        Jupyter_CRUD_Error: On update failure.
        SearchJupyterNotFoundError: If workspace does not exist (404).
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/datascience/v1/workspaces/{workspace_id}"

    res = await gd.get_data(
        url=url,
        method="PUT",
        auth=auth,
        body=config,
        context=context,
    )

    if return_raw:
        return res

    if res.status == 404:
        raise SearchJupyterNotFoundError(
            search_criteria=f"workspace_id: {workspace_id}", res=res
        )

    if not res.is_success:
        raise Jupyter_CRUD_Error(
            operation="update_config",
            workspace_id=workspace_id,
            message=f"Error updating workspace configuration for {workspace_id}",
            res=res,
        )

    return res

update_notebook_dataflow async

update_notebook_dataflow(
    auth: DomoJupyterAuth,
    workspace_id: str,
    notebook_id: str,
    body: dict[str, Any],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Update dataflow (scheduling) configuration for a notebook in a Jupyter workspace.

PUT .../services/domo/v1/workspaces/{workspace_id}/notebooks/{notebook_id}/dataflow

Parameters:

Name	Type	Description	Default
auth	DomoJupyterAuth	Jupyter authentication object.	required
workspace_id	str	Workspace UUID.	required
notebook_id	str	Notebook UUID.	required
body	dict[str, Any]	Request body (scheduling/dataflow payload). Shape is API-defined.	required
return_raw	bool	Return raw response without error checking.	False
context	RouteContext | None	Optional RouteContext.	None

Returns:

Type	Description
ResponseGetData	ResponseGetData with the API response.

Raises:

Type	Description
Jupyter_CRUD_Error	On failure.
SearchJupyterNotFoundError	On 404.

Source code in src/crew_dcs/routes/jupyter/content.py

@gd.route_function
@log_call(**_LOG_DECORATORS)
async def update_notebook_dataflow(
    auth: dmda.DomoJupyterAuth,
    workspace_id: str,
    notebook_id: str,
    body: dict[str, Any],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Update dataflow (scheduling) configuration for a notebook in a Jupyter workspace.

    ``PUT .../services/domo/v1/workspaces/{workspace_id}/notebooks/{notebook_id}/dataflow``

    Args:
        auth: Jupyter authentication object.
        workspace_id: Workspace UUID.
        notebook_id: Notebook UUID.
        body: Request body (scheduling/dataflow payload). Shape is API-defined.
        return_raw: Return raw response without error checking.
        context: Optional ``RouteContext``.

    Returns:
        ``ResponseGetData`` with the API response.

    Raises:
        Jupyter_CRUD_Error: On failure.
        SearchJupyterNotFoundError: On 404.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    dmda.test_is_jupyter_auth(auth)

    url = _build_jupyter_notebook_dataflow_url(auth, workspace_id, notebook_id)

    res = await gd.get_data(
        url=url,
        method="PUT",
        auth=auth,
        body=body,
        context=context,
    )

    if return_raw:
        return res

    if res.status == 404:
        raise SearchJupyterNotFoundError(
            search_criteria=f"workspace_id: {workspace_id}, notebook_id: {notebook_id}",
            res=res,
        )

    if not res.is_success:
        raise Jupyter_CRUD_Error(
            operation="update_notebook_dataflow",
            content_path=f"workspaces/{workspace_id}/notebooks/{notebook_id}/dataflow",
            res=res,
        )

    return res

Modules

• config
• content
• core
• exceptions