codeengine

codeengine ¶

CodeEngine Package

This package provides codeengine management functionality split across multiple modules for better organization.

Modules:

Name	Description
`exceptions`	Exception classes for codeengine operations
`core`	Core codeengine retrieval functions
`crud`	Create, read, update, delete operations
`upload`	Upload Python code to CodeEngine packages

CodeEnginePackageBuilder ¶

CodeEnginePackageBuilder()

Helper class for building CodeEngine packages.

Source code in src/crew_dcs/routes/codeengine/crud.py

def __init__(self):
    pass

CodeEngine_CRUD_Error ¶

CodeEngine_CRUD_Error(
    operation: str = "CRUD",
    entity_id: str | None = None,
    res: ResponseGetData | None = None,
    message: str | None = None,
    **kwargs
)

Bases: RouteError

Raised when codeengine create, update, or delete operations fail.

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

def __init__(
    self,
    operation: str = "CRUD",
    entity_id: str | None = None,
    res: rgd.ResponseGetData | None = None,
    message: str | None = None,
    **kwargs,
):
    if not message:
        message = f"CodeEngine {operation} operation failed"
    super().__init__(message=message, entity_id=entity_id, res=res, **kwargs)

CodeEngine_FunctionCallError ¶

CodeEngine_FunctionCallError(
    auth: DomoAuth | None = None,
    message: str | None = None,
    res: ResponseGetData | None = None,
    **kwargs
)

Bases: RouteError

Raised when function call parameters are invalid.

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

def __init__(
    self,
    auth: DomoAuth | None = None,
    message: str | None = None,
    res: rgd.ResponseGetData | None = None,
    **kwargs,
):
    # Extract domo_instance from auth if provided
    domo_instance = auth.domo_instance if auth else None
    super().__init__(
        message=message, domo_instance=domo_instance, res=res, **kwargs
    )

CodeEngine_GET_Error ¶

CodeEngine_GET_Error(
    entity_id: str | None = None,
    res: ResponseGetData | None = None,
    message: str | None = None,
    **kwargs
)

Bases: RouteError

Raised when codeengine retrieval operations fail.

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

def __init__(
    self,
    entity_id: str | None = None,
    res: rgd.ResponseGetData | None = None,
    message: str | None = None,
    **kwargs,
):
    if not message:
        if entity_id:
            message = f"Failed to retrieve codeengine package {entity_id}"
        else:
            message = "Failed to retrieve codeengine packages"

    super().__init__(message=message, entity_id=entity_id, res=res, **kwargs)

CodeEngine_InvalidPackageError ¶

CodeEngine_InvalidPackageError(
    message: str,
    auth: DomoAuth | None = None,
    res: ResponseGetData | None = None,
    **kwargs
)

Bases: RouteError

Raised when package validation fails.

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

def __init__(
    self,
    message: str,
    auth: DomoAuth | None = None,
    res: rgd.ResponseGetData | None = None,
    **kwargs,
):
    # Extract domo_instance from auth if provided
    domo_instance = auth.domo_instance if auth else None
    super().__init__(
        message=message, domo_instance=domo_instance, res=res, **kwargs
    )

CodeEngine_Package_Parts ¶

Bases: DomoEnumMixin, Enum

Enum for package parts parameter values.

SearchCodeEngineNotFoundError ¶

SearchCodeEngineNotFoundError(
    search_criteria: str,
    res: ResponseGetData | None = None,
    **kwargs
)

Bases: RouteError

Raised when codeengine search operations return no results.

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

def __init__(
    self,
    search_criteria: str,
    res: rgd.ResponseGetData | None = None,
    **kwargs,
):
    message = f"No codeengine packages found matching: {search_criteria}"
    super().__init__(
        message=message,
        res=res,
        **kwargs,
    )

create_codeengine_package `async` ¶

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

Create a new codeengine package.

Trap: a Google-style Args: docstring whose parameter description WRAPS across indented lines silently 400s here — _create_manifest_from_code's function-boundary extraction mis-slices the source. Use flowing-prose docstrings in any function you upload. (Bug is in crew-dcs manifest extraction, not the Domo API; not yet fixed.)

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`payload`	`dict`	Package data dictionary	required
`context`	`RouteContext \| None`	Route context (optional)	`None`
`return_raw`	`bool`	Return raw response without processing	`False`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing created package data

Raises:

Type	Description
`CodeEngine_CRUD_Error`	If package creation fails

Source code in src/crew_dcs/routes/codeengine/crud.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def create_codeengine_package(
    auth: DomoAuth,
    payload: dict,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Create a new codeengine package.

    Trap: a Google-style `Args:` docstring whose parameter description WRAPS
    across indented lines silently 400s here — `_create_manifest_from_code`'s
    function-boundary extraction mis-slices the source. Use flowing-prose
    docstrings in any function you upload. (Bug is in crew-dcs manifest
    extraction, not the Domo API; not yet fixed.)

    Args:
        auth: Authentication object
        payload: Package data dictionary
        context: Route context (optional)
        return_raw: Return raw response without processing

    Returns:
        ResponseGetData object containing created package data

    Raises:
        CodeEngine_CRUD_Error: If package creation fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    package_name = payload.get("name", "Unknown")
    package_version = payload.get("version", "Unknown")
    await logger.info(
        f"Creating CodeEngine package '{package_name}' version {package_version}"
    )

    url = f"https://{auth.domo_instance}.domo.com/api/codeengine/v2/packages"

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

    if return_raw:
        return res

    if not res.is_success:
        await logger.error(f"Failed to create package '{package_name}': {res.status}")
        raise CodeEngine_CRUD_Error(operation="create", res=res)

    package_id = res.response.get("id")
    await logger.info(
        f"Successfully created package '{package_name}' with ID {package_id}"
    )
    return res

delete_codeengine_package `async` ¶

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

Delete a codeengine package by ID.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`package_id`	`str`	Package identifier	required
`context`	`RouteContext \| None`	Route context (optional)	`None`
`return_raw`	`bool`	Return raw response without processing	`False`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing deletion result

Raises:

Type	Description
`CodeEngine_FunctionCallError`	If package_id is not provided
`CodeEngine_CRUD_Error`	If package deletion fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def delete_codeengine_package(
    auth: DomoAuth,
    package_id: str,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Delete a codeengine package by ID.

    Args:
        auth: Authentication object
        package_id: Package identifier
        context: Route context (optional)
        return_raw: Return raw response without processing

    Returns:
        ResponseGetData object containing deletion result

    Raises:
        CodeEngine_FunctionCallError: If package_id is not provided
        CodeEngine_CRUD_Error: If package deletion fails
    """
    from .crud import CodeEngine_CRUD_Error

    context = RouteContext.build_context(context=context, **context_kwargs)

    if not package_id:
        raise CodeEngine_FunctionCallError(
            message="Package ID must be provided.",
            auth=auth,
        )

    url = (
        f"https://{auth.domo_instance}.domo.com/api/codeengine/v2/packages/{package_id}"
    )

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

    if return_raw:
        return res

    if not res.is_success:
        raise CodeEngine_CRUD_Error(
            operation="delete",
            entity_id=package_id,
            res=res,
        )

    return res

delete_codeengine_package_version `async` ¶

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

Delete a specific version of a codeengine package.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`package_id`	`str`	Package identifier	required
`version`	`str`	Version string to delete	required
`context`	`RouteContext \| None`	Route context (optional)	`None`
`return_raw`	`bool`	Return raw response without processing	`False`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing deletion result

Raises:

Type	Description
`CodeEngine_FunctionCallError`	If package_id or version is not provided
`CodeEngine_CRUD_Error`	If version deletion fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def delete_codeengine_package_version(
    auth: DomoAuth,
    package_id: str,
    version: str,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Delete a specific version of a codeengine package.

    Args:
        auth: Authentication object
        package_id: Package identifier
        version: Version string to delete
        context: Route context (optional)
        return_raw: Return raw response without processing

    Returns:
        ResponseGetData object containing deletion result

    Raises:
        CodeEngine_FunctionCallError: If package_id or version is not provided
        CodeEngine_CRUD_Error: If version deletion fails
    """
    from .crud import CodeEngine_CRUD_Error

    context = RouteContext.build_context(context=context, **context_kwargs)

    if not package_id or not version:
        raise CodeEngine_FunctionCallError(
            message=f"Package ID {package_id or 'not provided'} and version {version or 'not provided'} must be provided.",
            auth=auth,
        )

    url = f"https://{auth.domo_instance}.domo.com/api/codeengine/v2/packages/{package_id}/versions/{version}"

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

    if return_raw:
        return res

    if not res.is_success:
        raise CodeEngine_CRUD_Error(
            operation="delete_version",
            entity_id=f"{package_id}/v{version}",
            res=res,
        )

    return res

deploy_codeengine_package `async` ¶

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

Deploy a specific codeengine package version.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`package_id`	`str`	Package identifier	required
`version`	`str`	Package version to deploy	required
`context`	`RouteContext \| None`	Route context (optional)	`None`
`return_raw`	`bool`	Return raw response without processing	`False`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing deployment result

Raises:

Type	Description
`CodeEngine_CRUD_Error`	If deployment fails

Source code in src/crew_dcs/routes/codeengine/crud.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def deploy_codeengine_package(
    auth: DomoAuth,
    package_id: str,
    version: str,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Deploy a specific codeengine package version.

    Args:
        auth: Authentication object
        package_id: Package identifier
        version: Package version to deploy
        context: Route context (optional)
        return_raw: Return raw response without processing

    Returns:
        ResponseGetData object containing deployment result

    Raises:
        CodeEngine_CRUD_Error: If deployment fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    await logger.info(f"Deploying CodeEngine package {package_id} version {version}")

    url = f"https://{auth.domo_instance}.domo.com/api/codeengine/v2/packages/{package_id}/versions/{version}/release"

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

    if return_raw:
        return res

    if not res.is_success:
        await logger.error(
            f"Failed to deploy package {package_id} v{version}: {res.status}"
        )
        raise CodeEngine_CRUD_Error(
            operation="deploy", entity_id=f"{package_id}/v{version}", res=res
        )

    await logger.info(f"Successfully deployed package {package_id} v{version}")
    return res

execute_codeengine_function `async` ¶

execute_codeengine_function(
    auth: DomoAuth,
    package_id: str,
    version: str,
    function_name: str,
    input_variables: dict,
    is_get_logs: bool = True,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs
)

Execute one function of a deployed package version.

input_variables must supply EVERY input the manifest declares — omitting one 400s with no hint which. Read the manifest first (get_codeengine_package_by_id_and_version(params={"parts": "functions"})).

Type traps (both fail at EXECUTION, not at package creation): - str | None raises TypeError: unsupported operand type(s) for | in Domo's own executor. Use typing.Optional[str], and pin it with # noqa: UP045 or ruff --fix rewrites it back to the broken form.

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def execute_codeengine_function(
    auth: DomoAuth,
    package_id: str,
    version: str,
    function_name: str,
    input_variables: dict,
    is_get_logs: bool = True,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs,
):
    """Execute one function of a deployed package version.

    `input_variables` must supply EVERY input the manifest declares — omitting
    one 400s with no hint which. Read the manifest first
    (`get_codeengine_package_by_id_and_version(params={"parts": "functions"})`).

    Type traps (both fail at EXECUTION, not at package creation):
      - `str | None` raises `TypeError: unsupported operand type(s) for |` in
        Domo's own executor. Use `typing.Optional[str]`, and pin it with
        `# noqa: UP045` or ruff `--fix` rewrites it back to the broken form.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/codeengine/v2/packages/{package_id}/versions/{version}/functions/{function_name}"

    res = await gd.get_data(
        method="POST",
        url=url,
        auth=auth,
        body={"inputVariables": input_variables, "settings": {"getLogs": is_get_logs}},
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success:
        raise CodeEngine_GET_Error(
            entity_id=f"{package_id}/v{version}/function/{function_name}", res=res
        )

    # Handle empty or non-dict responses
    if not res.response or not isinstance(res.response, dict):
        raise CodeEngine_FunctionCallError(
            message=f"Function execution returned invalid response: {res.response!r}",
            auth=auth,
            res=res,
        )

    if not res.response.get("status") == "SUCCESS":
        raise CodeEngine_FunctionCallError(
            message=f"Function execution failed with status {res.response.get('status')}",
            auth=auth,
            res=res,
        )

    response = res.response.pop("result", None)

    metadata = res.response

    if isinstance(response, dict):
        res.response = {**response, "_metadata": metadata}
    else:
        res.response = {"result": response, "_metadata": metadata}

    return res

generate_share_account_package ¶

generate_share_account_package(
    code: str,
    package_id: str,
    version: str,
    name: str,
    account_mappings: list[dict[str, Any]],
    environment: str = "LAMBDA",
    language: str = "PYTHON",
    functions: list[dict[str, Any]] | None = None,
) -> dict[str, Any]

Generate a CodeEngine package dictionary structure.

Same parameters as generate_package_json but returns a dict instead of JSON string. Useful when you need to manipulate the structure before serializing.

Returns:

Type	Description
`dict[str, Any]`	Dictionary in CodeEngine package format

Source code in src/crew_dcs/routes/codeengine/packages.py

def generate_share_account_package(
    code: str,
    package_id: str,
    version: str,
    name: str,
    account_mappings: list[dict[str, Any]],
    environment: str = "LAMBDA",
    language: str = "PYTHON",
    functions: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
    """
    Generate a CodeEngine package dictionary structure.

    Same parameters as generate_package_json but returns a dict instead of JSON string.
    Useful when you need to manipulate the structure before serializing.

    Returns:
        Dictionary in CodeEngine package format
    """
    return {
        "code": code,
        "environment": environment,
        "id": package_id,
        "language": language,
        "manifest": {
            "functions": functions or [],
            "configuration": {"accountsMapping": account_mappings},
        },
        "name": name,
        "version": version,
    }

get_codeengine_package_by_id `async` ¶

get_codeengine_package_by_id(
    auth: DomoAuth,
    package_id: str,
    params: dict | None = None,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs
) -> ResponseGetData

Retrieve a specific codeengine package by ID.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`package_id`	`str`	Package identifier	required
`params`	`dict \| None`	Query parameters (optional, defaults to {"parts": "versions"})	`None`
`context`	`RouteContext \| None`	Route context (optional)	`None`
`return_raw`	`bool`	Return raw response without processing	`False`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing package data

Raises:

Type	Description
`CodeEngine_FunctionCallError`	If package_id is not provided
`CodeEngine_GET_Error`	If package retrieval fails

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

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

    Args:
        auth: Authentication object
        package_id: Package identifier
        params: Query parameters (optional, defaults to {"parts": "versions"})
        context: Route context (optional)
        return_raw: Return raw response without processing

    Returns:
        ResponseGetData object containing package data

    Raises:
        CodeEngine_FunctionCallError: If package_id is not provided
        CodeEngine_GET_Error: If package retrieval fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    if not package_id:
        raise CodeEngine_FunctionCallError(
            message="Package ID must be provided.",
            auth=auth,
        )

    url = (
        f"https://{auth.domo_instance}.domo.com/api/codeengine/v2/packages/{package_id}"
    )

    params = params or {"parts": "versions,name"}

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

    if return_raw:
        return res

    if not res.is_success:
        raise CodeEngine_GET_Error(entity_id=package_id, res=res)

    return res

get_codeengine_package_by_id_and_version `async` ¶

get_codeengine_package_by_id_and_version(
    auth: DomoAuth,
    package_id: str,
    version: str,
    params: dict | None = None,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs
) -> ResponseGetData

Retrieve a specific codeengine package by ID and version.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`package_id`	`str`	Package identifier	required
`version`	`str`	Package version	required
`params`	`dict \| None`	Query parameters (optional, defaults to {"parts": "functions,code"})	`None`
`context`	`RouteContext \| None`	Route context (optional)	`None`
`return_raw`	`bool`	Return raw response without processing	`False`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing package version data

Raises:

Type	Description
`CodeEngine_FunctionCallError`	If package_id or version is not provided
`CodeEngine_GET_Error`	If package retrieval fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_codeengine_package_by_id_and_version(
    auth: DomoAuth,
    package_id: str,
    version: str,
    params: dict | None = None,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Retrieve a specific codeengine package by ID and version.

    Args:
        auth: Authentication object
        package_id: Package identifier
        version: Package version
        params: Query parameters (optional, defaults to {"parts": "functions,code"})
        context: Route context (optional)
        return_raw: Return raw response without processing

    Returns:
        ResponseGetData object containing package version data

    Raises:
        CodeEngine_FunctionCallError: If package_id or version is not provided
        CodeEngine_GET_Error: If package retrieval fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    if not package_id or not version:
        raise CodeEngine_FunctionCallError(
            message=f"Package ID {package_id or 'not provided'} and version {version or 'not provided'} must be provided.",
            auth=auth,
        )

    url = f"https://{auth.domo_instance}.domo.com/api/codeengine/v2/packages/{package_id}/versions/{version}"

    params = params or {"parts": "functions,code,name"}

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

    if return_raw:
        return res

    if not res.is_success:
        raise CodeEngine_GET_Error(entity_id=f"{package_id}/v{version}", res=res)

    return res

get_current_package_version `async` ¶

get_current_package_version(
    auth: DomoAuth,
    package_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> str

Get the current version of a CodeEngine package.

Convenience function that retrieves the package metadata and extracts the current version string from the versions array.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`package_id`	`str`	Package identifier	required
`context`	`RouteContext \| None`	Route context (optional)	`None`

Returns:

Name	Type	Description
`str`	`str`	Current package version (e.g., "1.0.0")

Raises:

Type	Description
`CodeEngine_GET_Error`	If package retrieval fails or version not found

Example

version = await get_current_package_version( ... auth=auth, ... package_id="b368d630-7ca5-4b8a-b4ec-f130cf312dc1" ... ) print(f"Current version: {version}") Current version: 1.2.3

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_current_package_version(
    auth: DomoAuth,
    package_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> str:
    """
    Get the current version of a CodeEngine package.

    Convenience function that retrieves the package metadata and extracts
    the current version string from the versions array.

    Args:
        auth: Authentication object
        package_id: Package identifier
        context: Route context (optional)

    Returns:
        str: Current package version (e.g., "1.0.0")

    Raises:
        CodeEngine_GET_Error: If package retrieval fails or version not found

    Example:
        >>> version = await get_current_package_version(
        ...     auth=auth,
        ...     package_id="b368d630-7ca5-4b8a-b4ec-f130cf312dc1"
        ... )
        >>> print(f"Current version: {version}")
        Current version: 1.2.3
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await get_codeengine_package_by_id(
        auth=auth,
        package_id=package_id,
        params={"parts": "versions"},
        context=context,
        return_raw=False,
    )

    # Extract version from response
    versions = res.response.get("versions", [])
    if not versions:
        raise CodeEngine_GET_Error(
            entity_id=package_id,
            res=res,
            message="No versions found for package",
        )

    # Return the last version (current version)
    current_version = versions[-1].get("version")
    if not current_version:
        raise CodeEngine_GET_Error(
            entity_id=package_id,
            res=res,
            message="Version string not found in package metadata",
        )

    return current_version

get_package_versions `async` ¶

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

Retrieve all versions of a codeengine package.

Each package can have one or many versions.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`package_id`	`str`	Package identifier	required
`context`	`RouteContext \| None`	Route context (optional)	`None`
`return_raw`	`bool`	Return raw response without processing	`False`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing package versions

Raises:

Type	Description
`CodeEngine_FunctionCallError`	If package_id is not provided
`CodeEngine_GET_Error`	If version retrieval fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_package_versions(
    auth: DomoAuth,
    package_id: str,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Retrieve all versions of a codeengine package.

    Each package can have one or many versions.

    Args:
        auth: Authentication object
        package_id: Package identifier
        context: Route context (optional)
        return_raw: Return raw response without processing

    Returns:
        ResponseGetData object containing package versions

    Raises:
        CodeEngine_FunctionCallError: If package_id is not provided
        CodeEngine_GET_Error: If version retrieval fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    if not package_id:
        raise CodeEngine_FunctionCallError(
            message="Package ID must be provided.",
            auth=auth,
        )

    url = f"https://{auth.domo_instance}.domo.com/api/codeengine/v2/packages/{package_id}/versions/"

    params = {"parts": "functions,code"}

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

    if return_raw:
        return res

    if not res.is_success:
        raise CodeEngine_GET_Error(entity_id=package_id, res=res)

    return res

get_packages `async` ¶

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

Retrieve all codeengine packages.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`context`	`RouteContext \| None`	Route context (optional)	`None`
`return_raw`	`bool`	Return raw response without processing	`False`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing package list

Raises:

Type	Description
`CodeEngine_GET_Error`	If package retrieval fails

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

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

    Args:
        auth: Authentication object
        context: Route context (optional)
        return_raw: Return raw response without processing

    Returns:
        ResponseGetData object containing package list

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

    url = f"https://{auth.domo_instance}.domo.com/api/codeengine/v2/packages"

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

    if return_raw:
        return res

    if not res.is_success:
        raise CodeEngine_GET_Error(res=res)

    return res

increment_version ¶

increment_version(version: str) -> str

Increment the version number.

Increments the last part of a dot-separated version string.

Parameters:

Name	Type	Description	Default
`version`	`str`	Version string (e.g., "1.0.0")	required

Returns:

Type	Description
`str`	Incremented version string (e.g., "1.0.1")

Source code in src/crew_dcs/routes/codeengine/crud.py

def increment_version(version: str) -> str:
    """
    Increment the version number.

    Increments the last part of a dot-separated version string.

    Args:
        version: Version string (e.g., "1.0.0")

    Returns:
        Incremented version string (e.g., "1.0.1")
    """
    parts = version.split(".")
    # Increment the last part
    parts[-1] = str(int(parts[-1]) + 1)
    return ".".join(parts)

share_accounts_with_package `async` ¶

share_accounts_with_package(
    auth: DomoAuth,
    package_id: str,
    account_mappings: list[dict[str, str | int]],
    is_set: bool = True,
    version_id: str | None = None,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    dry_run: bool = False,
    **context_kwargs
) -> ResponseGetData

Share Domo accounts with a CodeEngine package.

Updates the package manifest to include account mappings, allowing the package to access specific Domo accounts by alias.

⚠️ DESTRUCTIVE BY DEFAULT. is_set defaults to True, which REPLACES every existing account mapping on the package. Pass is_set=False to merge. There is no read-back, so a package that loses its mappings reports success.

dry_run=True simulates ONLY the destructive write. The reads this function needs (current version, package, version body) still execute, so the request body it shows you is the real one it would have sent. Nothing is written.

Check the result with res.is_dry_run — a simulated call still returns status=200, is_success=True, so is_success cannot tell the two apart.

See .agents/guides/access-and-sharing.md.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`package_id`	`str`	Package identifier	required
`account_mappings`	`list[dict[str, str \| int]]`	List of account mapping dicts with 'accountId' and 'alias' keys Example: [{"accountId": 92, "alias": "sdk_config"}]	required
`is_set`	`bool`	If True, replace all existing account mappings with new ones. If False, add new mappings to existing ones (default: True)	`True`
`context`	`RouteContext \| None`	Route context (optional)	`None`
`return_raw`	`bool`	Return raw response without processing	`False`
`dry_run`	`bool`	Simulate the write. Reads still run; the POST is not sent. Inspect `res.is_dry_run` and `res.response["body"]` to see what would have been submitted.	`False`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing updated package data

Raises:

Type	Description
`CodeEngine_CRUD_Error`	If account sharing fails

Example

Replace all account mappings¶

accounts = [ ... {"accountId": 92, "alias": "sdk_config"}, ... {"accountId": 189, "alias": "sdk_d2c"} ... ] res = await share_accounts_with_package( ... auth=auth, ... package_id="b368d630-7ca5-4b8a-b4ec-f130cf312dc1", ... account_mappings=accounts, ... is_set=True ... )

Add to existing account mappings¶

new_accounts = [{"accountId": 366, "alias": "sdk_audit"}] res = await share_accounts_with_package( ... auth=auth, ... package_id="b368d630-7ca5-4b8a-b4ec-f130cf312dc1", ... account_mappings=new_accounts, ... is_set=False ... )

Source code in src/crew_dcs/routes/codeengine/crud.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def share_accounts_with_package(
    auth: DomoAuth,
    package_id: str,
    account_mappings: list[dict[str, str | int]],
    is_set: bool = True,
    version_id: str | None = None,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    dry_run: bool = False,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Share Domo accounts with a CodeEngine package.

    Updates the package manifest to include account mappings, allowing the package
    to access specific Domo accounts by alias.

    ⚠️ DESTRUCTIVE BY DEFAULT. `is_set` defaults to True, which REPLACES every
    existing account mapping on the package. Pass `is_set=False` to merge. There
    is no read-back, so a package that loses its mappings reports success.

    `dry_run=True` simulates ONLY the destructive write. The reads this function
    needs (current version, package, version body) still execute, so the request
    body it shows you is the real one it would have sent. Nothing is written.

    Check the result with `res.is_dry_run` — a simulated call still returns
    `status=200, is_success=True`, so `is_success` cannot tell the two apart.

    See `.agents/guides/access-and-sharing.md`.

    Args:
        auth: Authentication object
        package_id: Package identifier
        account_mappings: List of account mapping dicts with 'accountId' and 'alias' keys
            Example: [{"accountId": 92, "alias": "sdk_config"}]
        is_set: If True, replace all existing account mappings with new ones.
            If False, add new mappings to existing ones (default: True)
        context: Route context (optional)
        return_raw: Return raw response without processing
        dry_run: Simulate the write. Reads still run; the POST is not sent.
            Inspect `res.is_dry_run` and `res.response["body"]` to see what
            would have been submitted.

    Returns:
        ResponseGetData object containing updated package data

    Raises:
        CodeEngine_CRUD_Error: If account sharing fails

    Example:
        >>> # Replace all account mappings
        >>> accounts = [
        ...     {"accountId": 92, "alias": "sdk_config"},
        ...     {"accountId": 189, "alias": "sdk_d2c"}
        ... ]
        >>> res = await share_accounts_with_package(
        ...     auth=auth,
        ...     package_id="b368d630-7ca5-4b8a-b4ec-f130cf312dc1",
        ...     account_mappings=accounts,
        ...     is_set=True
        ... )
        >>>
        >>> # Add to existing account mappings
        >>> new_accounts = [{"accountId": 366, "alias": "sdk_audit"}]
        >>> res = await share_accounts_with_package(
        ...     auth=auth,
        ...     package_id="b368d630-7ca5-4b8a-b4ec-f130cf312dc1",
        ...     account_mappings=new_accounts,
        ...     is_set=False
        ... )
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    # `dry_run` may arrive as this explicit parameter or on the context. Honour
    # either, but only for the WRITE: the reads below fetch the package code and
    # version that the request body is built from, and a simulated read returns
    # a stub with no "code"/"version" keys, which would blow up before the
    # caller ever saw the body they asked to preview.
    is_dry_run = bool(dry_run or context.dry_run)
    context = replace(context, dry_run=False)

    if is_dry_run:
        await logger.info(
            f"[DRY RUN] simulating account share for package {package_id}; "
            "reads will execute, the write will not be sent"
        )

    await logger.info(
        f"Sharing {len(account_mappings)} account mappings with package {package_id}"
    )

    # First, get the current package version
    if not version_id:
        await logger.debug(f"Retrieving current version for package {package_id}")
        version_id = await codeengine_routes.get_current_package_version(
            auth=auth, package_id=package_id, context=context
        )

    if not version_id:
        await logger.error(
            f"Could not determine current version for package {package_id}"
        )
        raise CodeEngine_CRUD_Error(
            operation="share_accounts",
            entity_id=package_id,
            message="Could not determine package version",
        )

    # Get the full version data including code
    res_p = await codeengine_routes.get_codeengine_package_by_id(
        auth=auth,
        package_id=package_id,
        # params={"parts": "code,environment,id,language,manifest,name,version"},
        context=context,
    )

    res_v = await codeengine_routes.get_codeengine_package_by_id_and_version(
        auth=auth,
        package_id=package_id,
        version=version_id,
        # params={"parts": "code,environment,id,language,manifest,name,version"},
        context=context,
    )

    if not is_set:
        # is_set=False MERGES: existing mappings are read and combined with the
        # new ones. (This comment previously read "Replace all existing mappings
        # with new ones" — inverted, and on the branch that does the opposite of
        # what it said. An agent picking `is_set` from the comment would choose
        # the destructive value, which is also the default.)
        await logger.debug("Merging new account mappings with existing mappings")
        existing_mappings = res_v.response["configuration"].get("accountsMapping", [])
        await logger.debug(
            f"Found {len(existing_mappings)} existing account mappings for version {version_id}"
        )

        # Build dict keyed by accountId for fast lookup
        existing_dict = {
            mapping.get("accountId"): mapping for mapping in existing_mappings
        }

        # Update with new mappings (overwrites if accountId exists, updates alias)
        for new_mapping in account_mappings:
            account_id = new_mapping.get("accountId")
            if account_id:
                existing_dict[account_id] = new_mapping

        # Convert back to list
        account_mappings = list(existing_dict.values())
        await logger.debug(f"Final account mappings count: {len(account_mappings)}")

    body = generate_share_account_package(
        package_id=res_p.response["id"],
        code=res_v.response["code"],
        version=res_v.response["version"],
        name=res_p.response["name"],
        environment=res_p.response["environment"],
        language=res_p.response["language"],
        functions=res_v.response["functions"],
        account_mappings=account_mappings,
    )

    url = f"https://{auth.domo_instance}.domo.com/api/codeengine/v2/packages"

    await logger.debug(
        f"Submitting account mappings for package {package_id} version {version_id}"
    )
    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        body=body,
        context=replace(context, dry_run=True) if is_dry_run else context,
    )

    if return_raw:
        return res

    if not res.is_success:
        await logger.error(
            f"Failed to share accounts with package {package_id}: {res.status}"
        )
        raise CodeEngine_CRUD_Error(
            operation="share_accounts",
            entity_id=package_id,
            res=res,
        )

    if res.is_dry_run:
        await logger.info(
            f"[DRY RUN] would have shared {len(account_mappings)} account mappings "
            f"with package {package_id}; nothing was written"
        )
        return res

    await logger.info(
        f"Successfully shared {len(account_mappings)} account mappings with package {package_id}"
    )
    return res

test_package_is_identical `async` ¶

test_package_is_identical(
    package_id: str,
    version: str,
    auth: DomoAuth,
    existing_package: dict | None = None,
    new_package: dict | None = None,
    new_code: str | None = None,
    params: dict | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> bool

Test if the code in a new package matches the existing package.

Parameters:

Name	Type	Description	Default
`package_id`	`str`	Package identifier	required
`version`	`str`	Package version	required
`auth`	`DomoAuth`	Authentication object	required
`existing_package`	`dict \| None`	Pre-fetched existing package data (optional)	`None`
`new_package`	`dict \| None`	New package data to compare (optional)	`None`
`new_code`	`str \| None`	New code to compare (optional)	`None`
`params`	`dict \| None`	Query parameters (optional)	`None`
`context`	`RouteContext \| None`	Route context (optional)	`None`
`**context_kwargs`		Additional context parameters	`{}`

Returns:

Type	Description
`bool`	True if the package code is identical, False otherwise

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

async def test_package_is_identical(
    package_id: str,
    version: str,
    auth: DomoAuth,
    existing_package: dict | None = None,
    new_package: dict | None = None,
    new_code: str | None = None,
    params: dict | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> bool:
    """
    Test if the code in a new package matches the existing package.

    Args:
        package_id: Package identifier
        version: Package version
        auth: Authentication object
        existing_package: Pre-fetched existing package data (optional)
        new_package: New package data to compare (optional)
        new_code: New code to compare (optional)
        params: Query parameters (optional)
        context: Route context (optional)
        **context_kwargs: Additional context parameters

    Returns:
        True if the package code is identical, False otherwise
    """
    context = RouteContext.build_context(context=context, **context_kwargs)
    _ = version

    existing_package = (
        existing_package
        or (
            await get_codeengine_package_by_id(
                auth=auth,
                package_id=package_id,
                params=params,
                context=context,
            )
        ).response
    )

    new_code = new_code or (new_package.get("code") if new_package else None)

    return existing_package.get("code") == new_code

test_package_is_released `async` ¶

test_package_is_released(
    package_id: str,
    version: str,
    auth: DomoAuth,
    existing_package: dict | None = None,
    params: dict | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> bool

Test if a package version is already released.

Parameters:

Name	Type	Description	Default
`package_id`	`str`	Package identifier	required
`version`	`str`	Package version	required
`auth`	`DomoAuth`	Authentication object	required
`existing_package`	`dict \| None`	Pre-fetched package data (optional)	`None`
`params`	`dict \| None`	Query parameters (optional)	`None`
`context`	`RouteContext \| None`	Route context (optional)	`None`
`**context_kwargs`		Additional context parameters	`{}`

Returns:

Type	Description
`bool`	True if the package is already released, False otherwise

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

async def test_package_is_released(
    package_id: str,
    version: str,
    auth: DomoAuth,
    existing_package: dict | None = None,
    params: dict | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> bool:
    """
    Test if a package version is already released.

    Args:
        package_id: Package identifier
        version: Package version
        auth: Authentication object
        existing_package: Pre-fetched package data (optional)
        params: Query parameters (optional)
        context: Route context (optional)
        **context_kwargs: Additional context parameters

    Returns:
        True if the package is already released, False otherwise
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    existing_package = (
        existing_package
        or (
            await get_codeengine_package_by_id_and_version(
                auth=auth,
                package_id=package_id,
                version=version,
                params=params,
                context=context,
            )
        ).response
    )

    return existing_package.get("released", False)

update_codeengine_package_owner `async` ¶

update_codeengine_package_owner(
    auth: DomoAuth,
    package_id: str,
    owner_id: int,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs
) -> ResponseGetData

Update the owner of a codeengine package.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`package_id`	`str`	Package identifier	required
`owner_id`	`int`	New owner user ID	required
`context`	`RouteContext \| None`	Route context (optional)	`None`
`return_raw`	`bool`	Return raw response without processing	`False`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing updated package data

Raises:

Type	Description
`CodeEngine_FunctionCallError`	If package_id is not provided
`CodeEngine_CRUD_Error`	If owner update fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def update_codeengine_package_owner(
    auth: DomoAuth,
    package_id: str,
    owner_id: int,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Update the owner of a codeengine package.

    Args:
        auth: Authentication object
        package_id: Package identifier
        owner_id: New owner user ID
        context: Route context (optional)
        return_raw: Return raw response without processing

    Returns:
        ResponseGetData object containing updated package data

    Raises:
        CodeEngine_FunctionCallError: If package_id is not provided
        CodeEngine_CRUD_Error: If owner update fails
    """
    from .crud import CodeEngine_CRUD_Error

    context = RouteContext.build_context(context=context, **context_kwargs)

    if not package_id:
        raise CodeEngine_FunctionCallError(
            message="Package ID must be provided.",
            auth=auth,
        )

    url = (
        f"https://{auth.domo_instance}.domo.com/api/codeengine/v2/packages/{package_id}"
    )

    body = {"owner": owner_id}

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

    if return_raw:
        return res

    if not res.is_success:
        raise CodeEngine_CRUD_Error(
            operation="update_owner",
            entity_id=package_id,
            res=res,
        )

    return res

upload_package_version `async` ¶

upload_package_version(
    auth: DomoAuth,
    package_id: str,
    code: str,
    is_new_version: bool = False,
    metadata: dict | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Upload Python code to a CodeEngine package.

This route function validates inputs, builds a manifest from the Python code, and deploys it to the specified CodeEngine package.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object with valid Domo credentials	required
`package_id`	`str`	ID of the CodeEngine package to update	required
`code`	`str`	Python source code string to deploy	required
`is_new_version`	`bool`	If True, create a new version. If False (default), update the current version.	`False`
`metadata`	`dict \| None`	Optional metadata overrides including: - runtime: Runtime version (default: "PYTHON_3_13") - accountsMapping: List of account mapping dicts - name: Package name override - description: Package description - handler_name: Name of the handler function	`None`
`return_raw`	`bool`	Return raw 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 containing: - package_id: The package ID - version: The version that was created/updated - status: "created" or "updated" - manifest: The generated manifest

Raises:

Type	Description
`CodeEngine_CRUD_Error`	If validation fails or deployment fails

Example

from crew_dcs.auth import DomoTokenAuth auth = DomoTokenAuth( ... domo_instance="mycompany", ... domo_access_token="your-token" ... ) code = ''' ... def process_data(items: list, limit: int = 10) -> dict: ... """Process items and return summary.""" ... return {"count": len(items[:limit])} ... ''' result = await upload_package_version( ... auth=auth, ... package_id="517ca12c-3459-4e66-b0bb-40f000720a84", ... code=code, ... is_new_version=False ... )

Note

Default runtime is Python 3.13 ("PYTHON_3_13")
Multi-file packages are not supported; code must be a single string
The first top-level function in the code is used as the handler unless handler_name is specified in metadata

Source code in src/crew_dcs/routes/codeengine/upload.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def upload_package_version(
    auth: DomoAuth,
    package_id: str,
    code: str,
    is_new_version: bool = False,
    metadata: dict | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Upload Python code to a CodeEngine package.

    This route function validates inputs, builds a manifest from the Python code,
    and deploys it to the specified CodeEngine package.

    Args:
        auth: Authentication object with valid Domo credentials
        package_id: ID of the CodeEngine package to update
        code: Python source code string to deploy
        is_new_version: If True, create a new version. If False (default),
            update the current version.
        metadata: Optional metadata overrides including:
            - runtime: Runtime version (default: "PYTHON_3_13")
            - accountsMapping: List of account mapping dicts
            - name: Package name override
            - description: Package description
            - handler_name: Name of the handler function
        return_raw: Return raw response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData containing:
            - package_id: The package ID
            - version: The version that was created/updated
            - status: "created" or "updated"
            - manifest: The generated manifest

    Raises:
        CodeEngine_CRUD_Error: If validation fails or deployment fails

    Example:
        >>> from crew_dcs.auth import DomoTokenAuth
        >>> auth = DomoTokenAuth(
        ...     domo_instance="mycompany",
        ...     domo_access_token="your-token"
        ... )
        >>> code = '''
        ... def process_data(items: list, limit: int = 10) -> dict:
        ...     \"\"\"Process items and return summary.\"\"\"
        ...     return {"count": len(items[:limit])}
        ... '''
        >>> result = await upload_package_version(
        ...     auth=auth,
        ...     package_id="517ca12c-3459-4e66-b0bb-40f000720a84",
        ...     code=code,
        ...     is_new_version=False
        ... )

    Note:
        - Default runtime is Python 3.13 ("PYTHON_3_13")
        - Multi-file packages are not supported; code must be a single string
        - The first top-level function in the code is used as the handler unless
          handler_name is specified in metadata
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    # Validate inputs
    if not package_id:
        raise CodeEngine_CRUD_Error(
            operation="upload",
            message="package_id is required",
        )

    if not code or not code.strip():
        raise CodeEngine_CRUD_Error(
            operation="upload",
            entity_id=package_id,
            message="code is required and cannot be empty",
        )

    # Import here to avoid circular imports
    from ...classes.DomoCodeEngine.CodeEngine import DomoCodeEngine_Package

    # Get package instance and deploy
    package = await DomoCodeEngine_Package.get_by_id(
        auth=auth,
        package_id=package_id,
        context=context,
    )

    try:
        result = await package.deploy_version(
            code=code,
            is_new_version=is_new_version,
            metadata=metadata,
            context=context,
        )

        # Build response
        res = rgd.ResponseGetData(
            status=200,
            response=result,
            is_success=True,
        )

        return res  # noqa: RET504

    except DomoError as e:
        # Wrap non-CodeEngine errors
        if not isinstance(e, CodeEngine_CRUD_Error):
            raise CodeEngine_CRUD_Error(
                operation="upload",
                entity_id=package_id,
                message=str(e),
            ) from e
        raise

upsert_codeengine_package_version `async` ¶

upsert_codeengine_package_version(
    auth: DomoAuth,
    payload: dict,
    version: str | None = None,
    auto_increment_version: bool = True,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs
) -> ResponseGetData

Create or update a codeengine package version.

If the package version exists and is released, optionally increment the version. If the package code is identical to existing, skip the update.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`payload`	`dict`	Package data dictionary	required
`version`	`str \| None`	Package version (optional, defaults to version in payload)	`None`
`auto_increment_version`	`bool`	Automatically increment version if deployed	`True`
`debug_prn`		Enable debug printing	required
`context`	`RouteContext \| None`	Route context (optional)	`None`
`return_raw`	`bool`	Return raw response without processing	`False`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing package data

Raises:

Type	Description
`CodeEngine_InvalidPackageError`	If package is already deployed and auto_increment is False
`CodeEngine_CRUD_Error`	If package creation/update fails

Source code in src/crew_dcs/routes/codeengine/crud.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def upsert_codeengine_package_version(
    auth: DomoAuth,
    payload: dict,
    version: str | None = None,
    auto_increment_version: bool = True,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Create or update a codeengine package version.

    If the package version exists and is released, optionally increment the version.
    If the package code is identical to existing, skip the update.

    Args:
        auth: Authentication object
        payload: Package data dictionary
        version: Package version (optional, defaults to version in payload)
        auto_increment_version: Automatically increment version if deployed
        debug_prn: Enable debug printing
        context: Route context (optional)
        return_raw: Return raw response without processing

    Returns:
        ResponseGetData object containing package data

    Raises:
        CodeEngine_InvalidPackageError: If package is already deployed and auto_increment is False
        CodeEngine_CRUD_Error: If package creation/update fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    package_id = payload.get("id")
    version = version or payload.get("version")

    try:
        existing_pkg = await codeengine_routes.get_codeengine_package_by_id_and_version(
            auth=auth,
            package_id=package_id,
            version=version,
            params={"parts": "code"},
            context=context,
        )
        if await codeengine_routes.test_package_is_released(
            existing_package=existing_pkg.response,
            package_id=package_id,
            version=version,
            auth=auth,
        ):
            if not auto_increment_version:
                raise CodeEngine_InvalidPackageError(
                    message=f"Package {package_id} v{version} already deployed",
                    auth=auth,
                )

            version = increment_version(version)

        if await codeengine_routes.test_package_is_identical(
            existing_package=existing_pkg.response,
            new_package=payload,
            package_id=package_id,
            version=version,
            auth=auth,
        ):
            await logger.debug(
                f"Package {package_id} v{version} is identical; skipping update."
            )
            return existing_pkg

    except CodeEngine_GET_Error:
        pass  # Not found, continue to create

    return await create_codeengine_package(
        auth=auth,
        payload=payload,
        context=context,
        return_raw=return_raw,
    )

upsert_package `async` ¶

upsert_package(
    auth: DomoAuth,
    payload: dict,
    check_different: bool = True,
    create_new_version: bool = False,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs
) -> ResponseGetData

Create or update a codeengine package.

If the package doesn't exist, create it. If it exists, update the version.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`payload`	`dict`	Package data dictionary	required
`check_different`	`bool`	If True, compare payload with existing package and skip update if identical (default: True)	`True`
`create_new_version`	`bool`	If True, force creation of a new version even if package exists (default: False)	`False`
`context`	`RouteContext \| None`	Route context (optional)	`None`
`return_raw`	`bool`	Return raw response without processing	`False`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing package data

Raises:

Type	Description
`CodeEngine_CRUD_Error`	If package creation/update fails

Source code in src/crew_dcs/routes/codeengine/crud.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def upsert_package(
    auth: DomoAuth,
    payload: dict,
    check_different: bool = True,
    create_new_version: bool = False,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Create or update a codeengine package.

    If the package doesn't exist, create it. If it exists, update the version.

    Args:
        auth: Authentication object
        payload: Package data dictionary
        check_different: If True, compare payload with existing package and skip
            update if identical (default: True)
        create_new_version: If True, force creation of a new version even if
            package exists (default: False)
        context: Route context (optional)
        return_raw: Return raw response without processing

    Returns:
        ResponseGetData object containing package data

    Raises:
        CodeEngine_CRUD_Error: If package creation/update fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    package_id = payload.get("id")
    version = payload.get("version")

    if not package_id:
        await logger.info("No Package ID found, creating new package...")

        return await create_codeengine_package(
            auth=auth,
            payload=payload,
            context=context,
            return_raw=return_raw,
        )

    # Try to get existing package
    await logger.debug(f"Checking if package {package_id} exists")
    try:
        existing_pkg = await codeengine_routes.get_codeengine_package_by_id(
            auth=auth,
            package_id=package_id,
            params={"parts": "code"},
            context=context,
        )
        await logger.debug(f"Found existing package {package_id}")
    except CodeEngine_GET_Error:
        # Package doesn't exist, create it
        await logger.info(f"Package {package_id} not found; creating new package")
        return await create_codeengine_package(
            auth=auth,
            payload=payload,
            context=context,
            return_raw=return_raw,
        )

    # If check_different is enabled, compare with existing package
    if check_different and version:
        await logger.debug(f"Comparing package {package_id} v{version} with existing")
        is_identical = await codeengine_routes.test_package_is_identical(
            existing_package=existing_pkg.response,
            new_package=payload,
            package_id=package_id,
            version=version,
            auth=auth,
            context=context,
        )

        if is_identical:
            await logger.info(
                f"Package {package_id} v{version} is identical to existing; skipping update."
            )
            return existing_pkg

        await logger.debug(
            f"Package {package_id} v{version} differs from existing; proceeding with update"
        )

    # If create_new_version is True, increment version before creating
    if create_new_version and version:
        version = increment_version(version)
        payload["version"] = version
        await logger.info(f"Creating new version {version} for package {package_id}")

    return await upsert_codeengine_package_version(
        payload=payload,
        auth=auth,
        auto_increment_version=not create_new_version,  # Don't auto-increment if we already did
        context=context,
        return_raw=return_raw,
    )

codeengine

codeengine ¶

CodeEnginePackageBuilder ¶

CodeEngine_CRUD_Error ¶

CodeEngine_FunctionCallError ¶

CodeEngine_GET_Error ¶

CodeEngine_InvalidPackageError ¶

CodeEngine_Package_Parts ¶

SearchCodeEngineNotFoundError ¶

create_codeengine_package async ¶

delete_codeengine_package async ¶

delete_codeengine_package_version async ¶

deploy_codeengine_package async ¶

execute_codeengine_function async ¶

generate_share_account_package ¶

get_codeengine_package_by_id async ¶

get_codeengine_package_by_id_and_version async ¶

get_current_package_version async ¶

get_package_versions async ¶

get_packages async ¶

increment_version ¶

share_accounts_with_package async ¶

Replace all account mappings¶

Add to existing account mappings¶

test_package_is_identical async ¶

test_package_is_released async ¶

update_codeengine_package_owner async ¶

upload_package_version async ¶

upsert_codeengine_package_version async ¶

upsert_package async ¶

Modules¶

create_codeengine_package `async` ¶

delete_codeengine_package `async` ¶

delete_codeengine_package_version `async` ¶

deploy_codeengine_package `async` ¶

execute_codeengine_function `async` ¶

get_codeengine_package_by_id `async` ¶

get_codeengine_package_by_id_and_version `async` ¶

get_current_package_version `async` ¶

get_package_versions `async` ¶

get_packages `async` ¶

share_accounts_with_package `async` ¶

test_package_is_identical `async` ¶

test_package_is_released `async` ¶

update_codeengine_package_owner `async` ¶

upload_package_version `async` ¶

upsert_codeengine_package_version `async` ¶

upsert_package `async` ¶