Jupyter

Jupyter

DomoJupyterWorkspace dataclass

DomoJupyterWorkspace(
    auth: DomoJupyterAuth,
    id: str,
    raw: dict,
    name: str,
    description: str,
    created_dt: datetime,
    updated_dt: datetime,
    owner: dict,
    cpu: str,
    memory: int,
    compute_tier: str | None = None,
    python_version: str | None = None,
    last_run_dt: datetime = None,
    instances: list[dict] = None,
    input_configuration: list[
        DomoJupyter_DataSource
    ] = list(),
    output_configuration: list[
        DomoJupyter_DataSource
    ] = list(),
    account_configuration: list[
        DomoJupyter_Account
    ] = list(),
    content: list[DomoJupyter_Content] = list(),
    fileshare_configuration: list[dict] = None,
    service_location: str = None,
    service_prefix: str = None,
)

Bases: DomoEntity

add_input_dataset async

add_input_dataset(
    domo_dataset: DomoDataset,
    domojupyter_ds: DomoJupyter_DataSource = None,
    domo_user: DomoUser = None,
    is_update_config: bool = True,
    debug_api: bool = False,
    session: AsyncClient = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Add a domo_dataset or domojupyter_dataset to workspace and conditionally update config.

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

async def add_input_dataset(
    self,
    domo_dataset: dmds.DomoDataset,
    domojupyter_ds: DomoJupyter_DataSource = None,
    domo_user: dmdu.DomoUser = None,
    is_update_config: bool = True,
    debug_api: bool = False,
    session: httpx.AsyncClient = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Add a domo_dataset or domojupyter_dataset to workspace and conditionally update config."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    if not domojupyter_ds and not isinstance(domo_dataset, dmds.DomoDataset):
        raise DJW_InvalidClass(
            message="must pass domo_dataset as class DomoDataset", cls_instance=self
        )

    if not domojupyter_ds:
        domojupyter_ds = DomoJupyter_DataSource(
            alias=domo_dataset.name, dataset_id=domo_dataset.id, dj_workspace=self
        )

    self.add_config_input_datasource(domojupyter_ds)

    if not is_update_config:
        return self.input_configuration

    async def do_share() -> None:
        nonlocal domo_user
        domo_user = domo_user or await dmdu.DomoUser.get_by_id(
            auth=self.auth,
            user_id=(await self.auth.who_am_i()).response["id"],
            debug_api=debug_api,
            session=context.session,
            context=context,
        )
        await domo_dataset.share(
            member=domo_user,
            auth=self.auth,
            debug_api=debug_api,
            session=context.session,
            context=context,
        )

    return await self._retry_config_after_share(
        context, do_share, "add_input_dataset"
    )

download_folder async

download_folder(
    target_folder: str | None = None,
    remote_folder: str = "",
    replace_target_folder: bool = True,
    ignore_folders: list[str] | None = None,
    included_filetypes: list[str] | None = None,
    exclude_hidden: bool = True,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None
) -> list[str]

Download workspace files under remote_folder into a local target_folder.

Paths under target_folder mirror the workspace relative to remote_folder (the remote prefix is stripped from on-disk layout).

Parameters:

Name	Type	Description	Default
target_folder	str | None	Local directory root for written files. Default: {domo_instance}/{workspace_name}.	None
remote_folder	str	Workspace-relative Jupyter path (empty string = entire tree).	''
replace_target_folder	bool	If True, replace target_folder before writing.	True
ignore_folders	list[str] | None	Path segments to exclude.	None
included_filetypes	list[str] | None	Extensions to include (e.g. ['.ipynb', '.py']); None = all.	None
exclude_hidden	bool	Skip remote files/folders whose Jupyter name starts with ..	True
debug_api	bool	Enable API request debugging.	False
session	AsyncClient | None	Optional httpx client session.	None
context	RouteContext | None	Optional RouteContext.	None

Returns:

Type	Description
list[str]	List of local file paths written.

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

async def download_folder(
    self,
    target_folder: str | None = None,
    remote_folder: str = "",
    replace_target_folder: bool = True,
    ignore_folders: list[str] | None = None,
    included_filetypes: list[str] | None = None,
    exclude_hidden: bool = True,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
) -> list[str]:
    """Download workspace files under ``remote_folder`` into a local ``target_folder``.

    Paths under ``target_folder`` mirror the workspace relative to ``remote_folder``
    (the remote prefix is stripped from on-disk layout).

    Args:
        target_folder: Local directory root for written files. Default:
            ``{domo_instance}/{workspace_name}``.
        remote_folder: Workspace-relative Jupyter path (empty string = entire tree).
        replace_target_folder: If True, replace ``target_folder`` before writing.
        ignore_folders: Path segments to exclude.
        included_filetypes: Extensions to include (e.g. ``['.ipynb', '.py']``); None = all.
        exclude_hidden: Skip remote files/folders whose Jupyter name starts with ``.``.
        debug_api: Enable API request debugging.
        session: Optional httpx client session.
        context: Optional RouteContext.

    Returns:
        List of local file paths written.
    """

    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=2,
    )

    dest = target_folder or f"{self.auth.domo_instance}/{self.name}"
    prefix_norm = _normalize_remote_folder(remote_folder)

    all_content = await self.get_content(
        remote_folder=remote_folder,
        ignore_folders=ignore_folders,
        included_filetypes=included_filetypes,
        exclude_hidden=exclude_hidden,
        debug_api=debug_api,
        context=context,
    )
    all_content = [
        content for content in all_content if content.file_type != "directory"
    ]
    # Trailing sep so upsert_folder replaces ``dest`` itself, not its parent (see utils/files upsert_folder).
    dest_norm = os.path.normpath(dest)
    defi.upsert_folder(f"{dest_norm}{os.sep}", replace_folder=replace_target_folder)

    written: list[str] = []
    for content in all_content:
        full_remote = _jupyter_item_remote_path(content)
        rel_disk = _disk_relative_under_remote(full_remote, prefix_norm)
        if not rel_disk:
            continue
        written.append(
            content.write_under_target(target_folder=dest, relative_path=rel_disk)
        )
    return written

get_account_configuration async

get_account_configuration(
    session: AsyncClient = None,
    debug_api: bool = False,
    is_suppress_errors: bool = False,
    is_use_default_account_class: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Load account configuration from raw data.

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

async def get_account_configuration(
    self: DomoJupyterWorkspace,
    session: httpx.AsyncClient = None,
    debug_api: bool = False,
    is_suppress_errors: bool = False,
    is_use_default_account_class: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Load account configuration from raw data."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=2,
        **context_kwargs,
    )

    if self.raw.get("accountConfiguration"):
        from ...base.exceptions import DomoError
        from ...routes.account.exceptions import AccountNoMatchError
        from .. import DomoAccount as dmac

        # Fetch accounts first
        async def fetch_account(ac, context):
            try:
                return await dmac.DomoAccount.get_by_id(
                    auth=self.auth,
                    account_id=ac["account_id"],
                    is_use_default_account_class=is_use_default_account_class,
                    is_suppress_no_config=is_suppress_errors,
                    context=context,
                    debug_api=debug_api,
                )
            except (AccountNoMatchError, DomoError):
                return None

        accounts = await dmce.gather_with_concurrency(
            *[
                fetch_account(ac, context=context)
                for ac in self.raw["accountConfiguration"]
            ],
            n=10,
        )

        # Create DomoJupyter_Account instances with fetched accounts
        self.account_configuration = [
            DomoJupyter_Account.from_dict(
                obj=ac,
                dj_workspace=self,
                domo_account=account,
            )
            for ac, account in zip(self.raw["accountConfiguration"], accounts)  # noqa: B905
        ]
    return self.account_configuration

get_input_configuration async

get_input_configuration(
    session: AsyncClient = None,
    debug_api: bool = False,
    is_suppress_errors: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Load input configuration from raw data.

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

async def get_input_configuration(
    self: DomoJupyterWorkspace,
    session: httpx.AsyncClient = None,
    debug_api: bool = False,
    is_suppress_errors: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Load input configuration from raw data."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=2,
        **context_kwargs,
    )

    if self.raw.get("inputConfiguration"):
        import crew_dcs.routes.dataset as dataset_route

        # Fetch datasets first
        async def fetch_dataset(ic, context):
            try:
                return await dmds.DomoDataset.get_by_id(
                    auth=self.auth,
                    dataset_id=ic["dataSourceId"],
                    is_suppress_no_config=is_suppress_errors,
                    context=context,
                    debug_api=debug_api,
                )
            except dataset_route.DatasetNotFoundError:
                return None

        datasets = await dmce.gather_with_concurrency(
            *[
                fetch_dataset(ic, context=context)
                for ic in self.raw["inputConfiguration"]
            ],
            n=10,
        )

        # Create DomoJupyter_DataSource instances with fetched datasets
        self.input_configuration = [
            DomoJupyter_DataSource.from_dict(
                obj=ic,
                dj_workspace=self,
                domo_dataset=dataset,
            )
            for ic, dataset in zip(self.raw["inputConfiguration"], datasets)  # noqa: B905
        ]
    return self.input_configuration

get_notebook_dataflow async

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

Get dataflow (scheduling) config for a notebook. notebook_id is the UUID used by the dataflow API (e.g. from browser network tab or Domo).

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

async def get_notebook_dataflow(
    self,
    notebook_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get dataflow (scheduling) config for a notebook. notebook_id is the UUID used by the dataflow API (e.g. from browser network tab or Domo)."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    return await jupyter_routes.get_notebook_dataflow(
        auth=self.auth,
        workspace_id=self.id,
        notebook_id=notebook_id,
        return_raw=return_raw,
        context=context,
    )

get_notebooks async

get_notebooks(
    remote_folder: str = "",
    ignore_folders: list[str] | None = None,
    included_filetypes: list[str] | None = None,
    exclude_hidden: bool = True,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[JupyterNotebook]

Return only notebook content items as JupyterNotebook instances. Fetches content if needed.

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

async def get_notebooks(
    self,
    remote_folder: str = "",
    ignore_folders: list[str] | None = None,
    included_filetypes: list[str] | None = None,
    exclude_hidden: bool = True,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[JupyterNotebook]:
    """Return only notebook content items as JupyterNotebook instances. Fetches content if needed."""
    if not self.content or remote_folder or ignore_folders is not None:
        await self.get_content(
            remote_folder=remote_folder,
            ignore_folders=ignore_folders,
            included_filetypes=included_filetypes or [".ipynb"],
            exclude_hidden=exclude_hidden,
            context=context,
            **context_kwargs,
        )
    return [
        JupyterNotebook.from_content(c)
        for c in self.content
        if c.file_type == "notebook" or c.name.endswith(".ipynb")
    ]

get_output_configuration async

get_output_configuration(
    session: AsyncClient = None,
    debug_api: bool = False,
    is_suppress_errors: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Load output configuration from raw data.

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

async def get_output_configuration(
    self: DomoJupyterWorkspace,
    session: httpx.AsyncClient = None,
    debug_api: bool = False,
    is_suppress_errors: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Load output configuration from raw data."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=2,
        **context_kwargs,
    )

    if self.raw.get("outputConfiguration"):
        import crew_dcs.routes.dataset as dataset_route

        # Fetch datasets first
        async def fetch_dataset(oc, context):
            try:
                return await dmds.DomoDataset.get_by_id(
                    auth=self.auth,
                    dataset_id=oc["dataSourceId"],
                    is_suppress_no_config=is_suppress_errors,
                    session=session,
                    debug_api=debug_api,
                    context=context,
                )
            except dataset_route.DatasetNotFoundError:
                return None

        datasets = await dmce.gather_with_concurrency(
            *[
                fetch_dataset(oc, context=context)
                for oc in self.raw["outputConfiguration"]
            ],
            n=10,
        )

        # Create DomoJupyter_DataSource instances with fetched datasets
        self.output_configuration = [
            DomoJupyter_DataSource.from_dict(
                obj=oc,
                dj_workspace=self,
                domo_dataset=dataset,
            )
            for oc, dataset in zip(self.raw["outputConfiguration"], datasets)  # noqa: B905
        ]
    return self.output_configuration

prune_vestigial_accounts async

prune_vestigial_accounts(
    *, context: RouteContext | None = None, **context_kwargs
) -> list[DomoJupyter_Account]

Remove accounts that no longer exist (deleted in Domo but still in workspace config). Returns removed items.

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

async def prune_vestigial_accounts(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoJupyter_Account]:
    """Remove accounts that no longer exist (deleted in Domo but still in workspace config). Returns removed items."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    removed: list[DomoJupyter_Account] = []
    for account in list(self.account_configuration):
        await account.get_account(
            is_suppress_errors=True,
            session=context.session,
            debug_api=context.debug_api if context else False,
        )
        if not account.is_exists:
            self._remove_config(
                "account_configuration",
                lambda a: a.account_id == account.account_id,  # noqa: B023
            )
            removed.append(account)
    if removed:
        await self.update_config(context=context)
    return removed

remove_config_account async

remove_config_account(
    alias_or_id: str | int,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Remove an account from workspace config by alias or account_id. Persists via update_config.

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

async def remove_config_account(
    self,
    alias_or_id: str | int,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Remove an account from workspace config by alias or account_id. Persists via update_config."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    key = str(alias_or_id)
    removed = self._remove_config(
        "account_configuration",
        lambda a: a.alias == key or str(a.account_id) == key,
    )
    if removed:
        await self.update_config(context=context)
    return removed

remove_config_input_datasource async

remove_config_input_datasource(
    alias_or_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Remove an input datasource from workspace config by alias or dataset_id. Persists via update_config.

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

async def remove_config_input_datasource(
    self,
    alias_or_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Remove an input datasource from workspace config by alias or dataset_id. Persists via update_config."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    removed = self._remove_config(
        "input_configuration",
        lambda d: d.alias == alias_or_id or d.dataset_id == alias_or_id,
    )
    if removed:
        await self.update_config(context=context)
    return removed

remove_config_output_datasource async

remove_config_output_datasource(
    alias_or_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Remove an output datasource from workspace config by alias or dataset_id. Persists via update_config.

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

async def remove_config_output_datasource(
    self,
    alias_or_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Remove an output datasource from workspace config by alias or dataset_id. Persists via update_config."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    removed = self._remove_config(
        "output_configuration",
        lambda d: d.alias == alias_or_id or d.dataset_id == alias_or_id,
    )
    if removed:
        await self.update_config(context=context)
    return removed

sync_content async

sync_content(
    target_folder: str,
    remote_folder: str = "",
    direction: str = "pull",
    ignore_folders: list[str] | None = None,
    included_filetypes: list[str] | None = None,
    exclude_hidden: bool = True,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[str] | tuple[list[str], list[str]]

Sync workspace content with a local folder.

Parameters:

Name	Type	Description	Default
target_folder	str	Local directory path.	required
remote_folder	str	Workspace-relative Jupyter prefix for both pull and push.	''
direction	str	"pull" (download from workspace), "push" (upload to workspace), or "both" (pull then push).	'pull'
ignore_folders	list[str] | None	Folder names to exclude.	None
included_filetypes	list[str] | None	File extensions to include; None = all.	None
exclude_hidden	bool	Skip dot-prefixed names on pull and push.	True
debug_api	bool	Enable API debugging.	False
session	AsyncClient | None	Optional httpx session.	None
context	RouteContext | None	Optional RouteContext.	None

Returns:

Type	Description
list[str] | tuple[list[str], list[str]]	For "pull", list of local file paths written; for "push", list of content paths updated; for "both", (pull_paths, push_paths).

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

async def sync_content(
    self,
    target_folder: str,
    remote_folder: str = "",
    direction: str = "pull",
    ignore_folders: list[str] | None = None,
    included_filetypes: list[str] | None = None,
    exclude_hidden: bool = True,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[str] | tuple[list[str], list[str]]:
    """Sync workspace content with a local folder.

    Args:
        target_folder: Local directory path.
        remote_folder: Workspace-relative Jupyter prefix for both pull and push.
        direction: "pull" (download from workspace), "push" (upload to workspace), or "both" (pull then push).
        ignore_folders: Folder names to exclude.
        included_filetypes: File extensions to include; None = all.
        exclude_hidden: Skip dot-prefixed names on pull and push.
        debug_api: Enable API debugging.
        session: Optional httpx session.
        context: Optional RouteContext.

    Returns:
        For "pull", list of local file paths written; for "push", list of content paths updated; for "both", (pull_paths, push_paths).
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=2,
        **context_kwargs,
    )
    if direction == "pull":
        return await self.download_folder(
            target_folder=target_folder,
            remote_folder=remote_folder,
            ignore_folders=ignore_folders,
            included_filetypes=included_filetypes,
            exclude_hidden=exclude_hidden,
            debug_api=debug_api,
            session=session,
            context=context,
            **context_kwargs,
        )
    if direction == "push":
        return await self.upload_folder(
            target_folder=target_folder,
            remote_folder=remote_folder,
            ignore_folders=ignore_folders,
            included_filetypes=included_filetypes,
            exclude_hidden=exclude_hidden,
            debug_api=debug_api,
            session=session,
            context=context,
            **context_kwargs,
        )
    if direction == "both":
        pull_paths = await self.download_folder(
            target_folder=target_folder,
            remote_folder=remote_folder,
            ignore_folders=ignore_folders,
            included_filetypes=included_filetypes,
            exclude_hidden=exclude_hidden,
            debug_api=debug_api,
            session=session,
            context=context,
            **context_kwargs,
        )
        push_paths = await self.upload_folder(
            target_folder=target_folder,
            remote_folder=remote_folder,
            ignore_folders=ignore_folders,
            included_filetypes=included_filetypes,
            exclude_hidden=exclude_hidden,
            debug_api=debug_api,
            session=session,
            context=context,
            **context_kwargs,
        )
        return (pull_paths, push_paths)
    raise ValueError(
        f'direction must be "pull", "push", or "both"; got {direction!r}'
    )

update_auth_to_jupyter_auth

update_auth_to_jupyter_auth(jupyter_token: str)

interacting with jupyter notebook content requires an additional authentication against the jupyter server (jupyter_token) currently we must manually scrape the token from the browser

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

def update_auth_to_jupyter_auth(self, jupyter_token: str):
    """
    interacting with jupyter notebook content requires an additional authentication against the jupyter server (jupyter_token)
    currently we must manually scrape the token from the browser
    """

    self._update_auth_params()

    if isinstance(self.auth, DomoJupyterAuth):
        pass

    elif isinstance(self.auth, DomoFullAuth):
        self.auth = DomoJupyterFullAuth.convert_auth(
            auth=self.auth,
            service_location=self.service_location,
            jupyter_token=jupyter_token,
            service_prefix=self.service_prefix,
        )

    elif isinstance(self.auth, DomoTokenAuth):
        self.auth = DomoJupyterTokenAuth.convert_auth(
            auth=self.auth,
            service_location=self.service_location,
            jupyter_token=jupyter_token,
            service_prefix=self.service_prefix,
        )
    return self.auth

update_notebook_dataflow async

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

Update dataflow (scheduling) config for a notebook.

notebook_id is the UUID used by the dataflow API (e.g. from browser network tab or Domo when available). body is the API-defined payload.

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

async def update_notebook_dataflow(
    self,
    notebook_id: str,
    body: dict[str, Any],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Update dataflow (scheduling) config for a notebook.

    notebook_id is the UUID used by the dataflow API (e.g. from browser network tab
    or Domo when available). body is the API-defined payload.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)
    return await jupyter_routes.update_notebook_dataflow(
        auth=self.auth,
        workspace_id=self.id,
        notebook_id=notebook_id,
        body=body,
        return_raw=return_raw,
        context=context,
    )

upload_folder async

upload_folder(
    target_folder: str,
    remote_folder: str = "",
    ignore_folders: list[str] | None = None,
    included_filetypes: list[str] | None = None,
    exclude_hidden: bool = True,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[str]

Upload files from a local target_folder into the workspace under remote_folder.

Each file's path relative to target_folder is joined with remote_folder on the server.

Parameters:

Name	Type	Description	Default
target_folder	str	Local directory to read from.	required
remote_folder	str	Workspace-relative prefix (empty = workspace root).	''
ignore_folders	list[str] | None	Path segments to skip under target_folder.	None
included_filetypes	list[str] | None	Extensions to include; None = all.	None
exclude_hidden	bool	Skip local files and directories whose name starts with ..	True
debug_api	bool	Enable API debugging.	False
session	AsyncClient | None	Optional httpx session.	None
context	RouteContext | None	Optional RouteContext.	None

Returns:

Type	Description
list[str]	List of remote content paths that were created or updated.

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

async def upload_folder(  # noqa: C901
    self,
    target_folder: str,
    remote_folder: str = "",
    ignore_folders: list[str] | None = None,
    included_filetypes: list[str] | None = None,
    exclude_hidden: bool = True,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[str]:
    """Upload files from a local ``target_folder`` into the workspace under ``remote_folder``.

    Each file's path relative to ``target_folder`` is joined with ``remote_folder`` on the server.

    Args:
        target_folder: Local directory to read from.
        remote_folder: Workspace-relative prefix (empty = workspace root).
        ignore_folders: Path segments to skip under ``target_folder``.
        included_filetypes: Extensions to include; None = all.
        exclude_hidden: Skip local files and directories whose name starts with ``.``.
        debug_api: Enable API debugging.
        session: Optional httpx session.
        context: Optional RouteContext.

    Returns:
        List of remote content paths that were created or updated.
    """
    import json as _json

    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=2,
        **context_kwargs,
    )
    src = os.path.abspath(target_folder)
    ignore_folders = ignore_folders or []
    included_filetypes = included_filetypes or []
    prefix_norm = _normalize_remote_folder(remote_folder)

    file_entries: list[tuple[str, str]] = []
    dirs_set: set[str] = set()
    for root, dirs, files in os.walk(src):
        if exclude_hidden:
            dirs[:] = [d for d in dirs if not d.startswith(".")]
        rel_root = os.path.relpath(root, src)
        if rel_root == ".":
            rel_root = ""
        for name in files:
            if exclude_hidden and name.startswith("."):
                continue
            if ignore_folders and any(
                ig in os.path.normpath(os.path.join(rel_root, name)).split(os.sep)
                for ig in ignore_folders
            ):
                continue
            if included_filetypes and not any(
                name.endswith(ext) for ext in included_filetypes
            ):
                continue
            rel_path = os.path.join(rel_root, name)
            rel_posix = rel_path.replace(os.sep, "/")
            remote_path = _join_remote_path(prefix_norm, rel_posix)
            file_entries.append((remote_path, os.path.join(root, name)))
            parts = remote_path.split("/")[:-1]
            for i in range(len(parts)):
                dirs_set.add("/".join(parts[: i + 1]))

    for dir_path in sorted(dirs_set, key=lambda p: (p.count("/"), p)):
        try:
            res = await jupyter_routes.get_jupyter_content(
                auth=self.auth,
                content_path=dir_path,
                include_content=False,
                return_raw=True,
                context=context,
            )
        except Exception:  # noqa: BLE001
            res = None
        if not res or not res.is_success:
            await jupyter_routes.create_folder(
                auth=self.auth,
                folder_path=dir_path,
                context=context,
            )

    updated: list[str] = []
    for remote_path, full_path in file_entries:
        with open(full_path, encoding="utf-8") as f:
            raw = f.read()
        new_content = _json.loads(raw) if remote_path.endswith(".ipynb") else raw

        try:
            res = await jupyter_routes.get_jupyter_content(
                auth=self.auth,
                content_path=remote_path,
                include_content=False,
                return_raw=True,
                context=context,
            )
        except Exception:  # noqa: BLE001
            res = None
        if res and res.is_success:
            await jupyter_routes.update_jupyter_file(
                auth=self.auth,
                content_path=remote_path,
                new_content=new_content,
                context=context,
            )
        else:
            await jupyter_routes.create_jupyter_obj(
                auth=self.auth,
                content_path=remote_path,
                new_content=new_content,
                context=context,
            )
        updated.append(remote_path)
    return updated