content

content

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"}

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_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

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

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_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