CodeEngine

CodeEngine ¶

DomoCodeEngine Package and Version Classes

This module provides classes for interacting with Domo CodeEngine packages including package management, version control, and code deployment.

Class Hierarchy¶

The CodeEngine classes follow the DomoEntity pattern:

DomoCodeEngine_Package (DomoEntity): Represents a CodeEngine package
Contains multiple versions
Manages package metadata (name, description, language)
Provides methods for creation, retrieval, updates, and deployment
DomoCodeEngine_PackageVersion (DomoSubEntity): Represents a specific version
Belongs to a parent DomoCodeEngine_Package
Contains source code and function definitions
Inherits authentication from parent package
May contain a CodeEngineManifest for Python packages
DomoCodeEngine_Packages (DomoManager): Manages collections of packages
Provides search and retrieval methods
Returns lists of DomoCodeEngine_Package instances

Relationships¶

classDiagram
    class DomoEntity {
        +str id
        +DomoAuth auth
        +dict raw
        +display_url() str
        +from_dict() DomoEntity
        +get_by_id() DomoEntity
    }

    class DomoSubEntity {
        +DomoEntity parent
        +auth (from parent)
    }

    class DomoManager {
        +DomoAuth auth
        +get() list
    }

    class DomoCodeEngine_Package {
        +str id
        +str name
        +str language
        +str current_version
        +list~DomoCodeEngine_PackageVersion~ versions
        +create() DomoCodeEngine_Package
        +get_by_id() DomoCodeEngine_Package
        +upsert() DomoCodeEngine_Package
        +deploy_version() dict
        +_create_manifest_from_code() CodeEngineManifest
        +_build_payload() dict
    }

    class DomoCodeEngine_PackageVersion {
        +str package_id
        +str version
        +str code
        +CodeEngineManifest Manifest
        +from_dict() DomoCodeEngine_PackageVersion
        +get_by_id_and_version() DomoCodeEngine_PackageVersion
        +download_source_code() str
    }

    class DomoCodeEngine_Packages {
        +get() list~DomoCodeEngine_Package~
        +search_by_name() list~DomoCodeEngine_Package~
    }

    class CodeEngineManifest {
        +list~CodeEngineManifest_Function~ functions
        +dict configuration
        +from_python_string() CodeEngineManifest
    }

    DomoEntity <|-- DomoCodeEngine_Package
    DomoSubEntity <|-- DomoCodeEngine_PackageVersion
    DomoManager <|-- DomoCodeEngine_Packages
    DomoCodeEngine_Package "1" *-- "many" DomoCodeEngine_PackageVersion : contains
    DomoCodeEngine_PackageVersion "1" *-- "0..1" CodeEngineManifest : has
    DomoCodeEngine_Packages "1" *-- "many" DomoCodeEngine_Package : manages
    DomoCodeEngine_Package ..> CodeEngineManifest : creates via _create_manifest_from_code

Usage Example¶

from crew_dcs.auth import DomoTokenAuth
from crew_dcs.classes.DomoCodeEngine import DomoCodeEngine_Package

auth = DomoTokenAuth(domo_instance="...", domo_access_token="...")

# Create a new package from Python code
code = "def process_data(input_data: list, limit: int = 100) -> dict:\n    return {'processed': len(input_data[:limit])}"

package = await DomoCodeEngine_Package.create(
    auth=auth,
    name="Data Processor",
    code=code,
    version="1.0.0"
)

# Get the current version
current_version = await package.get_current_version()
print(f"Current version: {current_version.version}")

# Deploy new code to existing package
result = await package.deploy_version(
    code=updated_code,
    is_new_version=True
)

# Download source code
file_path = await current_version.download_source_code()

Classes:

Name	Description
`DomoCodeEngine_Package`	Main package entity class
`DomoCodeEngine_Packages`	Manager class for package collections
`DomoCodeEngine_PackageVersion`	Package version subentity

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

DomoCodeEngine_ConfigError ¶

DomoCodeEngine_ConfigError(
    cls_instance=None,
    package_id: str | None = None,
    version: str | None = None,
    message: str | None = None,
    domo_instance: str | None = None,
)

Bases: ClassError

Exception raised when CodeEngine configuration is invalid.

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

def __init__(
    self,
    cls_instance=None,
    package_id: str | None = None,
    version: str | None = None,
    message: str | None = None,
    domo_instance: str | None = None,
):
    full_message = f"version {version} | {message}" if version else message
    super().__init__(
        cls_instance=cls_instance,
        entity_id=package_id,
        message=full_message,
        domo_instance=domo_instance,
    )

DomoCodeEngine_Package `dataclass` ¶

DomoCodeEngine_Package(
    auth: DomoAuth,
    id: str,
    raw: dict,
    name: str,
    description: str,
    language: str,
    environment: str,
    availability: str,
    owner_id: int,
    created: datetime,
    last_modified: datetime,
    functions: list,
    current_version: str | None = None,
    versions: (
        list[DomoCodeEngine_PackageVersion] | None
    ) = None,
    owner: list[DomoUser] | None = None,
)

Bases: DomoEntity

Domo CodeEngine Package entity.

Represents a CodeEngine package with versions and associated metadata. Packages contain executable code functions that can be deployed in Domo.

Attributes:

Name	Type	Description
`id`	`str`	Unique package identifier
`auth`	`DomoAuth`	Authentication object
`raw`	`dict`	Raw API response data
`name`	`str`	Package display name
`description`	`str`	Package description
`language`	`str`	Programming language (PYTHON or JAVASCRIPT)
`environment`	`str`	Execution environment
`availability`	`str`	Package availability status
`owner_id`	`int`	ID of the package owner
`created`	`datetime`	Package creation datetime
`last_modified`	`datetime`	Last modification datetime
`functions`	`list`	List of function definitions
`current_version`	`str \| None`	Latest version number
`versions`	`list[DomoCodeEngine_PackageVersion] \| None`	List of DomoCodeEngine_PackageVersion instances
`owner`	`list[DomoUser] \| None`	DomoUser instance of the package owner

Relationships

Package → Versions: One package contains many versions (one-to-many)
Package → Manager: DomoCodeEngine_Packages manages collections of packages
Version → Manifest: Each version may contain a CodeEngineManifest (for Python packages)

Class Constants

DEFAULT_RUNTIME: Default runtime version ("PYTHON_3_13") DEFAULT_ENVIRONMENT: Default execution environment ("LAMBDA") DEFAULT_LANGUAGE: Default programming language ("PYTHON")

Key Methods

create(): Create a new package from Python code
deploy_version(): Deploy code to this package (updates or creates new version)
_create_manifest_from_code(): Internal method to create manifest from Python code
_build_payload(): Internal method to build API payload

Example

package = await DomoCodeEngine_Package.create( ... auth=auth, ... name="My Package", ... code="def hello(): return 'world'" ... ) result = await package.deploy_version( ... code="def hello(): return 'updated'", ... is_new_version=True ... )

describe `property` ¶

describe: dict

Quick introspection — package description + function names and docstrings.

Use this to answer "does this CodeEngine library already have a function that does X?" before building a new one.

Note: For Python packages, functions are parsed from the code via Manifest (with full docstrings, input args, output types). For JavaScript packages, functions come from the API version response.

Returns:

Type	Description
`dict`	dict with name, description, language, current_version, function_count,
`dict`	and a list of function summaries (name, display_name, description/docstring,
`dict`	is_private, input names).

display_url `property` ¶

display_url: str

Generate the URL to display this package in the Domo interface.

associate_accounts `async` ¶

associate_accounts(
    account_mappings: list[dict[str, str | int]],
    is_set: bool = True,
    version_id: str | None = None,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs
)

Associate Domo accounts with this package so its functions can call codeengine.get_account().

Account associations map a Domo account (by ID) to an alias. Inside a deployed Code Engine function, codeengine.get_account(alias) looks up credentials by that alias — the function never receives raw secrets as text inputs. The Domo UI has deprecated the account-association picker, but the mapping remains a required part of the Code Engine architecture: without it, get_account() has nothing to resolve. Call this after deploying a package (i.e. after current_version is set) to wire up the aliases the function code expects.

Parameters:

Name	Type	Description	Default
`account_mappings`	`list[dict[str, str \| int]]`	List of dicts with 'accountId' and 'alias' keys. Example: [{"accountId": 13, "alias": "domo"}, {"accountId": 16, "alias": "github"}]	required
`is_set`	`bool`	If True, replace all existing account mappings with new ones. If False, merge new mappings into existing ones (default: True).	`True`
`version_id`	`str \| None`	Package version to associate accounts with. Defaults to this package's current_version.	`None`
`context`	`RouteContext \| None`	Optional RouteContext for API call configuration.	`None`
`return_raw`	`bool`	If True, return the raw ResponseGetData.	`False`
`**context_kwargs`		Additional context parameters.	`{}`

Returns:

Type	Description
	ResponseGetData for the account-sharing request.

Example

package = await DomoCodeEngine_Package.upsert(auth=auth, name="Domo Docs Sync", code=code) await package.associate_accounts([ ... {"accountId": 16, "alias": "github"}, ... {"accountId": 13, "alias": "domo"}, ... ])

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

async def associate_accounts(
    self,
    account_mappings: list[dict[str, str | int]],
    is_set: bool = True,
    version_id: str | None = None,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs,
):
    """Associate Domo accounts with this package so its functions can call codeengine.get_account().

    Account associations map a Domo account (by ID) to an alias. Inside a deployed
    Code Engine function, ``codeengine.get_account(alias)`` looks up credentials by
    that alias — the function never receives raw secrets as text inputs. The Domo UI
    has deprecated the account-association picker, but the mapping remains a required
    part of the Code Engine architecture: without it, ``get_account()`` has nothing to
    resolve. Call this after deploying a package (i.e. after ``current_version`` is set)
    to wire up the aliases the function code expects.

    Args:
        account_mappings: List of dicts with 'accountId' and 'alias' keys.
            Example: [{"accountId": 13, "alias": "domo"}, {"accountId": 16, "alias": "github"}]
        is_set: If True, replace all existing account mappings with new ones.
            If False, merge new mappings into existing ones (default: True).
        version_id: Package version to associate accounts with. Defaults to
            this package's current_version.
        context: Optional RouteContext for API call configuration.
        return_raw: If True, return the raw ResponseGetData.
        **context_kwargs: Additional context parameters.

    Returns:
        ResponseGetData for the account-sharing request.

    Example:
        >>> package = await DomoCodeEngine_Package.upsert(auth=auth, name="Domo Docs Sync", code=code)
        >>> await package.associate_accounts([
        ...     {"accountId": 16, "alias": "github"},
        ...     {"accountId": 13, "alias": "domo"},
        ... ])
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    return await codeengine_routes.share_accounts_with_package(
        auth=self.auth,
        package_id=self.id,
        account_mappings=account_mappings,
        is_set=is_set,
        version_id=version_id or self.current_version,
        context=context,
        return_raw=return_raw,
    )

create `async` `classmethod` ¶

create(
    auth: DomoAuth,
    name: str,
    code: str,
    version: str = "1.0.0",
    metadata: dict | None = None,
    *,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: AsyncClient | None = None,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoCodeEngine_Package

Create a new CodeEngine package from Python source code.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object with access to the target Domo instance.	required
`name`	`str`	Display name for the new package.	required
`code`	`str`	Python source code string to upload.	required
`version`	`str`	Initial version string (default: "1.0.0").	`'1.0.0'`
`metadata`	`dict \| None`	Optional overrides for manifest / package configuration. Supported keys: - runtime: Runtime version (default: "PYTHON_3_13") - accountsMapping: List of account mapping dicts - description: Package description - environment: Execution environment (default: "LAMBDA") - language: Language identifier (default: "PYTHON")	`None`
`debug_api`	`bool`	Enable API debugging for underlying calls.	`False`
`debug_num_stacks_to_drop`	`int`	Stack frames to drop in logging.	`2`
`session`	`AsyncClient \| None`	Optional httpx session for connection reuse.	`None`

Returns:

Type	Description
`DomoCodeEngine_Package`	DomoCodeEngine_Package instance for the created package.

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

@classmethod
async def create(
    cls,
    auth: DomoAuth,
    name: str,
    code: str,
    version: str = "1.0.0",
    metadata: dict | None = None,
    *,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: httpx.AsyncClient | None = None,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoCodeEngine_Package:
    """Create a new CodeEngine package from Python source code.

    Args:
        auth: Authentication object with access to the target Domo instance.
        name: Display name for the new package.
        code: Python source code string to upload.
        version: Initial version string (default: "1.0.0").
        metadata: Optional overrides for manifest / package configuration. Supported keys:
            - runtime: Runtime version (default: "PYTHON_3_13")
            - accountsMapping: List of account mapping dicts
            - description: Package description
            - environment: Execution environment (default: "LAMBDA")
            - language: Language identifier (default: "PYTHON")
        debug_api: Enable API debugging for underlying calls.
        debug_num_stacks_to_drop: Stack frames to drop in logging.
        session: Optional httpx session for connection reuse.

    Returns:
        DomoCodeEngine_Package instance for the created package.
    """
    metadata = metadata or {}

    await logger.info(
        f"Creating CodeEngine package '{name}' with initial version {version}"
    )

    manifest = cls._create_manifest_from_code(code=code, metadata=metadata)
    await logger.debug(
        f"Generated manifest with {len(manifest.functions)} functions"
    )

    functions = [fn.to_dict() for fn in manifest.functions]

    environment = metadata.get("environment", cls.DEFAULT_ENVIRONMENT)
    language = metadata.get("language", cls.DEFAULT_LANGUAGE)
    description = metadata.get("description")
    initial_version = metadata.get("version", version)

    payload: dict = {
        "code": code,
        "environment": environment,
        "language": language,
        "manifest": {
            "functions": functions,
            "configuration": manifest.configuration,
        },
        "name": name,
        "version": initial_version,
    }

    if description is not None:
        payload["description"] = description

    await logger.debug(f"Submitting package creation payload for '{name}'")

    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        **context_kwargs,
    )

    res = await codeengine_routes.create_codeengine_package(
        auth=auth,
        payload=payload,
        context=context,
        return_raw=False,
    )

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

    # Re-fetch the package to get full data including versions
    # (the create API response doesn't include version information)
    return await cls.get_by_id(
        auth=auth,
        package_id=package_id,
        context=context,
    )

delete `async` ¶

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

Delete this CodeEngine package.

Parameters:

Name	Type	Description	Default
`return_raw`	`bool`	If True, return raw API response	`False`
`context`	`RouteContext \| None`	Optional RouteContext for API call configuration	`None`
`**context_kwargs`		Additional context parameters	`{}`

Returns:

Type	Description
	ResponseGetData containing deletion result

Raises:

Type	Description
`CodeEngine_CRUD_Error`	If deletion fails

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

async def delete(
    self,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Delete this CodeEngine package.

    Args:
        return_raw: If True, return raw API response
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        ResponseGetData containing deletion result

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

    await logger.info(f"Deleting CodeEngine package {self.id}")
    res = await codeengine_routes.delete_codeengine_package(
        auth=self.auth,
        package_id=self.id,
        context=context,
    )

    if return_raw:
        return res

    await logger.info(f"Successfully deleted package {self.id}")
    return res

deploy_release `async` `classmethod` ¶

deploy_release(
    auth: DomoAuth,
    package_id: str,
    version: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Deploy (release) a specific package version.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`package_id`	`str`	Package identifier	required
`version`	`str`	Version string to deploy	required
`return_raw`	`bool`	If True, return raw API response	`False`
`context`	`RouteContext \| None`	Optional RouteContext for API call configuration	`None`
`**context_kwargs`		Additional context parameters	`{}`

Returns:

Type	Description
	ResponseGetData if return_raw=True, otherwise the deployment result

Raises:

Type	Description
`CodeEngine_CRUD_Error`	If deployment fails

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

@classmethod
async def deploy_release(
    cls,
    auth: DomoAuth,
    package_id: str,
    version: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Deploy (release) a specific package version.

    Args:
        auth: Authentication object
        package_id: Package identifier
        version: Version string to deploy
        return_raw: If True, return raw API response
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        ResponseGetData if return_raw=True, otherwise the deployment result

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

    await logger.info(
        f"Deploying release for package {package_id} version {version}"
    )
    res = await codeengine_routes.deploy_codeengine_package(
        auth=auth,
        package_id=package_id,
        version=version,
        context=context,
    )

    if return_raw:
        return res

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

deploy_version `async` ¶

deploy_version(
    code: str,
    is_new_version: bool = False,
    metadata: dict | None = None,
    context: RouteContext | None = None,
    **context_kwargs
) -> dict

Deploy Python code to this package.

Creates a manifest from the code and deploys it to this package. Supports either updating the current version or creating a new version.

Parameters:

Name	Type	Description	Default
`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`
`debug_api`		Enable API debugging	required
`session`		Optional httpx session for connection reuse	required

Returns:

Type	Description
`dict`	dict containing: - package_id: The package ID - version: The version that was created/updated - status: "created" or "updated" - response: Raw API response data

Raises:

Type	Description
`CodeEngine_GET_Error`	If package retrieval fails
`CodeEngine_CRUD_Error`	If deployment fails

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

async def deploy_version(
    self,
    code: str,
    is_new_version: bool = False,
    metadata: dict | None = None,
    context: RouteContext | None = None,
    **context_kwargs,
) -> dict:
    """Deploy Python code to this package.

    Creates a manifest from the code and deploys it to this package.
    Supports either updating the current version or creating a new version.

    Args:
        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
        debug_api: Enable API debugging
        session: Optional httpx session for connection reuse

    Returns:
        dict containing:
            - package_id: The package ID
            - version: The version that was created/updated
            - status: "created" or "updated"
            - response: Raw API response data

    Raises:
        CodeEngine_GET_Error: If package retrieval fails
        CodeEngine_CRUD_Error: If deployment fails
    """

    context = RouteContext.build_context(context=context, **context_kwargs)
    metadata = metadata or {}

    await logger.info(f"Deploying version to package {self.id}")

    # Build manifest from code
    handler_name = metadata.get("handler_name")
    await logger.debug(
        f"Building manifest from code (handler: {handler_name or 'auto'})"
    )
    manifest = self._create_manifest_from_code(
        code=code,
        handler_name=handler_name,
        metadata=metadata,
    )

    # Determine version
    current_version = self.current_version or "1.0.0"
    if is_new_version:
        from ...routes.codeengine import increment_version

        version = increment_version(current_version)
        await logger.info(
            f"Creating new version {version} (from {current_version})"
        )
    else:
        version = current_version
        await logger.info(f"Updating existing version {version}")

    # Build payload
    payload = self._build_payload(
        code=code,
        manifest=manifest,
        version=version,
        metadata=metadata,
    )

    # Deploy using upsert route
    from ...routes.codeengine import upsert_package

    await logger.debug(f"Submitting deployment payload for version {version}")
    res = await upsert_package(
        auth=self.auth,
        payload=payload,
        create_new_version=is_new_version,
        context=context,
    )

    await logger.info(
        f"Successfully deployed version {version} to package {self.id}"
    )
    return {
        "package_id": self.id,
        "version": version,
        "status": "created" if is_new_version else "updated",
        "response": res.response,
    }

describe_async `async` ¶

describe_async() -> dict

Async version of describe — fetches current version if functions not populated.

Use this when the package was retrieved from a list (where functions aren't included) and you need the full function details.

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

async def describe_async(self) -> dict:
    """Async version of describe — fetches current version if functions not populated.

    Use this when the package was retrieved from a list (where functions aren't
    included) and you need the full function details.
    """
    functions = self.functions or []
    if not functions and self.current_version:
        try:
            cv = await self.get_current_version()
            # Python packages: use Manifest functions (parsed from code)
            if cv.Manifest and cv.Manifest.functions:
                functions = [fn.to_dict() for fn in cv.Manifest.functions]
            # JS packages: use API functions from version response
            elif cv.functions:
                functions = cv.functions
        except DomoError:
            # get_current_version() raises CodeEngine_GET_Error (RouteError)
            # or DomoCodeEngine_ConfigError (ClassError) — both DomoError
            # subclasses. Keep empty functions if version fetch fails.
            pass

    return {
        "name": self.name,
        "description": self.description,
        "language": self.language,
        "current_version": self.current_version,
        "function_count": len(functions),
        "functions": [
            {
                "name": fn.get("name"),
                "display_name": fn.get("displayName"),
                "description": fn.get("description"),
                "is_private": fn.get("isPrivate", False),
                "inputs": [inp.get("name") for inp in fn.get("inputs", [])],
            }
            for fn in functions
        ],
    }

from_dict `classmethod` ¶

from_dict(auth: DomoAuth, obj: dict[str, Any])

Create DomoCodeEngine_Package from API response dictionary.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`obj`	`dict[str, Any]`	API response dictionary	required

Returns:

Type	Description
	DomoCodeEngine_Package instance

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

@classmethod
def from_dict(cls, auth: DomoAuth, obj: dict[str, Any]):
    """Create DomoCodeEngine_Package from API response dictionary.

    Args:
        auth: Authentication object
        obj: API response dictionary

    Returns:
        DomoCodeEngine_Package instance
    """
    package_id = str(obj.get("id"))
    language = obj.get("language")

    # Create package first (without versions) so we can reference it as parent
    package = cls(
        auth=auth,
        id=package_id,
        name=obj.get("name"),
        description=obj.get("description"),
        language=language,
        environment=obj.get("environment"),
        availability=obj.get("availability"),
        owner_id=obj.get("owner"),
        versions=None,  # Will set after creating versions
        created=convert_string_to_datetime(obj.get("createdOn")),
        last_modified=convert_string_to_datetime(obj.get("updatedOn")),
        functions=obj.get("functions", []),
        raw=obj,
    )

    # Parse versions if present, now that we have the parent package
    versions = []
    if obj.get("versions"):
        versions = [
            DomoCodeEngine_PackageVersion.from_dict(
                auth=auth,
                obj=version,
                package_id=package_id,
                language=language,
                parent=package,
            )
            for version in obj.get("versions", [])
        ]
        package.versions = versions

    # Re-set current version now that versions are populated
    # (versions is None during __post_init__, so _set_current_version
    # was a no-op when called from __post_init__)
    package._set_current_version()

    return package

get_by_id `async` `classmethod` ¶

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

Retrieve a CodeEngine package by ID.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`package_id`	`str`	Package identifier	required
`params`	`dict \| None`	Optional query parameters	`None`
`return_raw`	`bool`	If True, return raw API response	`False`
`debug_api`		Enable API debugging	required
`debug_num_stacks_to_drop`		Stack frames to drop in logging	required
`session`		Optional httpx session	required
`context`	`RouteContext \| None`	Optional RouteContext for API call configuration	`None`
`**context_kwargs`		Additional context parameters	`{}`

Returns:

Type	Description
	DomoCodeEngine_Package instance or ResponseGetData if return_raw=True

Raises:

Type	Description
`CodeEngine_GET_Error`	If package retrieval fails

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

@classmethod
async def get_by_id(
    cls,
    auth: DomoAuth,
    package_id: str,
    params: dict | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Retrieve a CodeEngine package by ID.

    Args:
        auth: Authentication object
        package_id: Package identifier
        params: Optional query parameters
        return_raw: If True, return raw API response
        debug_api: Enable API debugging
        debug_num_stacks_to_drop: Stack frames to drop in logging
        session: Optional httpx session
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        DomoCodeEngine_Package instance or ResponseGetData if return_raw=True

    Raises:
        CodeEngine_GET_Error: If package retrieval fails
    """

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

    await logger.debug(f"Fetching CodeEngine package {package_id}")
    res = await codeengine_routes.get_codeengine_package_by_id(
        auth=auth,
        package_id=package_id,
        params=params,
        context=context,
    )

    if return_raw:
        return res

    await logger.info(f"Successfully retrieved package {package_id}")
    return cls.from_dict(auth=auth, obj=res.response)

get_current_version `async` ¶

get_current_version(
    *, context: RouteContext | None = None, **context_kwargs
) -> DomoCodeEngine_PackageVersion

Get the current (latest) version of this package.

Parameters:

Name	Type	Description	Default
`context`	`RouteContext \| None`	Optional RouteContext for API call configuration	`None`
`**context_kwargs`		Additional context parameters (debug_api, session, debug_num_stacks_to_drop, etc.)	`{}`

Returns:

Type	Description
`DomoCodeEngine_PackageVersion`	DomoCodeEngine_PackageVersion instance

Raises:

Type	Description
`DomoCodeEngine_ConfigError`	If no current version found

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

async def get_current_version(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoCodeEngine_PackageVersion:
    """Get the current (latest) version of this package.

    Args:
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters (debug_api, session, debug_num_stacks_to_drop, etc.)

    Returns:
        DomoCodeEngine_PackageVersion instance

    Raises:
        DomoCodeEngine_ConfigError: If no current version found
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    if not self.current_version:
        await logger.error(f"No current version found for package {self.id}")
        raise DomoCodeEngine_ConfigError(
            package_id=self.id,
            version=None,
            message="No current version found for the package",
            domo_instance=self.auth.domo_instance,
        )

    await logger.debug(
        f"Fetching current version {self.current_version} for package {self.id}"
    )
    domo_version = await DomoCodeEngine_PackageVersion.get_by_id_and_version(
        auth=self.auth,
        package_id=self.id,
        version=self.current_version,
        language=self.language,
        context=context,
    )

    await logger.info(
        f"Retrieved current version {self.current_version} for package {self.id}"
    )
    return domo_version

get_entity_by_id `async` `classmethod` ¶

get_entity_by_id(auth: DomoAuth, entity_id: str, **kwargs)

Fetch a CodeEngine package by its ID.

This method is used by the base DomoEntity class for entity resolution.

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

@classmethod
async def get_entity_by_id(cls, auth: DomoAuth, entity_id: str, **kwargs):
    """Fetch a CodeEngine package by its ID.

    This method is used by the base DomoEntity class for entity resolution.
    """
    return await cls.get_by_id(auth=auth, package_id=entity_id, **kwargs)

get_owner `async` ¶

get_owner(
    *, context: RouteContext | None = None, **context_kwargs
) -> DomoUser | None

Get the owner (user) of this package.

Parameters:

Name	Type	Description	Default
`context`	`RouteContext \| None`	Optional RouteContext for API call configuration	`None`
`**context_kwargs`		Additional context parameters (debug_api, session, debug_num_stacks_to_drop, etc.)	`{}`

Returns:

Type	Description
`DomoUser \| None`	DomoUser instance or None if owner_id not set

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

async def get_owner(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> dmdu.DomoUser | None:
    """Get the owner (user) of this package.

    Args:
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters (debug_api, session, debug_num_stacks_to_drop, etc.)

    Returns:
        DomoUser instance or None if owner_id not set
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    if not self.owner_id:
        await logger.debug(f"No owner_id set for package {self.id}")
        return None

    await logger.debug(
        f"Fetching owner for package {self.id} (owner_id: {self.owner_id})"
    )
    self.owner = await dmdu.DomoUser.get_by_id(
        auth=self.auth,
        user_id=str(self.owner_id),
        context=context,
    )

    await logger.info(f"Retrieved owner for package {self.id}")
    return self.owner

run_function `async` `classmethod` ¶

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

Execute a CodeEngine function.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`package_id`	`str`	Package identifier	required
`version`	`str`	Package version	required
`function_name`	`str`	Name of the function to execute	required
`input_variables`	`dict`	Dictionary of input variables for the function	required
`is_get_logs`	`bool`	Whether to retrieve execution logs (default: True)	`True`
`return_raw`	`bool`	If True, return raw API response	`False`
`context`	`RouteContext \| None`	Optional RouteContext for API call configuration	`None`
`**context_kwargs`		Additional context parameters	`{}`

Returns:

Type	Description
	ResponseGetData if return_raw=True, otherwise the function execution result

Raises:

Type	Description
`CodeEngine_GET_Error`	If function execution fails
`CodeEngine_FunctionCallError`	If function call is invalid

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

@classmethod
async def run_function(
    cls,
    auth: DomoAuth,
    package_id: str,
    version: str,
    function_name: str,
    input_variables: dict,
    is_get_logs: bool = True,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Execute a CodeEngine function.

    Args:
        auth: Authentication object
        package_id: Package identifier
        version: Package version
        function_name: Name of the function to execute
        input_variables: Dictionary of input variables for the function
        is_get_logs: Whether to retrieve execution logs (default: True)
        return_raw: If True, return raw API response
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        ResponseGetData if return_raw=True, otherwise the function execution result

    Raises:
        CodeEngine_GET_Error: If function execution fails
        CodeEngine_FunctionCallError: If function call is invalid
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    await logger.info(
        f"Running function '{function_name}' on package {package_id} v{version}"
    )
    res = await codeengine_routes.execute_codeengine_function(
        auth=auth,
        package_id=package_id,
        version=version,
        function_name=function_name,
        input_variables=input_variables,
        is_get_logs=is_get_logs,
        context=context,
    )

    if return_raw:
        return res

    await logger.info(
        f"Successfully executed function '{function_name}' on package {package_id} v{version}"
    )
    return res

update_owner `async` ¶

update_owner(
    owner_id: int,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Update the owner of this CodeEngine package.

Parameters:

Name	Type	Description	Default
`owner_id`	`int`	New owner user ID	required
`return_raw`	`bool`	If True, return raw API response	`False`
`context`	`RouteContext \| None`	Optional RouteContext for API call configuration	`None`
`**context_kwargs`		Additional context parameters	`{}`

Returns:

Type	Description
	ResponseGetData if return_raw=True, otherwise the updated package

Raises:

Type	Description
`CodeEngine_CRUD_Error`	If owner update fails

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

async def update_owner(
    self,
    owner_id: int,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Update the owner of this CodeEngine package.

    Args:
        owner_id: New owner user ID
        return_raw: If True, return raw API response
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        ResponseGetData if return_raw=True, otherwise the updated package

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

    await logger.info(f"Updating owner to {owner_id} for package {self.id}")
    res = await codeengine_routes.update_codeengine_package_owner(
        auth=self.auth,
        package_id=self.id,
        owner_id=owner_id,
        context=context,
    )

    if return_raw:
        return res

    self.owner_id = owner_id
    await logger.info(f"Successfully updated owner for package {self.id}")
    return self

upsert `async` `classmethod` ¶

upsert(
    auth: DomoAuth,
    name: str,
    code: str,
    *,
    search_existing: bool = True,
    create_if_missing: bool = True,
    is_new_version: bool = True,
    version: str = "1.0.0",
    metadata: dict | None = None,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoCodeEngine_Package

Create or update a CodeEngine package from Python source code.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object.	required
`name`	`str`	Logical package name to search for / create.	required
`code`	`str`	Python source to deploy.	required
`search_existing`	`bool`	If True, search by name before creating.	`True`
`create_if_missing`	`bool`	If True, create a new package when none exists.	`True`
`is_new_version`	`bool`	Passed to deploy_version for existing packages.	`True`
`version`	`str`	Initial version to use when creating a new package.	`'1.0.0'`
`metadata`	`dict \| None`	Optional manifest / package metadata overrides.	`None`
`debug_api`		Enable API debugging for underlying calls.	required
`debug_num_stacks_to_drop`		Stack frames to drop in logging.	required
`session`		Optional httpx session for connection reuse.	required

Returns:

Type	Description
`DomoCodeEngine_Package`	DomoCodeEngine_Package for the updated or created package.

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

@classmethod
async def upsert(
    cls,
    auth: DomoAuth,
    name: str,
    code: str,
    *,
    search_existing: bool = True,
    create_if_missing: bool = True,
    is_new_version: bool = True,
    version: str = "1.0.0",
    metadata: dict | None = None,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoCodeEngine_Package:
    """Create or update a CodeEngine package from Python source code.

    Args:
        auth: Authentication object.
        name: Logical package name to search for / create.
        code: Python source to deploy.
        search_existing: If True, search by name before creating.
        create_if_missing: If True, create a new package when none exists.
        is_new_version: Passed to deploy_version for existing packages.
        version: Initial version to use when creating a new package.
        metadata: Optional manifest / package metadata overrides.
        debug_api: Enable API debugging for underlying calls.
        debug_num_stacks_to_drop: Stack frames to drop in logging.
        session: Optional httpx session for connection reuse.

    Returns:
        DomoCodeEngine_Package for the updated or created package.
    """
    context = RouteContext.build_context(
        context=context,
        **context_kwargs,
    )

    metadata = metadata or {}

    await logger.info(f"Upserting CodeEngine package '{name}'")

    # 1) Try to find an existing package by name
    existing_pkg: DomoCodeEngine_Package | None = None
    if search_existing:
        await logger.debug(f"Searching for existing package with name '{name}'")
        manager = DomoCodeEngine_Packages(auth=auth)
        try:
            matches = await manager.search_by_name(
                name=name,
                context=context,
                **context_kwargs,
            )
        except SearchCodeEngineNotFoundError:
            matches = []

        # Prefer exact name match if multiple results come back
        exact_matches = [
            pkg for pkg in matches if (pkg.name or "").lower() == name.lower()
        ]
        if exact_matches:
            existing_pkg = exact_matches[0]
            await logger.debug(f"Found exact name match for package '{name}'")
        elif matches:
            # Fallback: first partial match
            existing_pkg = matches[0]
            await logger.debug(f"Found partial name match for package '{name}'")
        else:
            await logger.debug(f"No existing package found with name '{name}'")

    # 2) If we found an existing package, deploy a new or updated version
    if existing_pkg is not None:
        await logger.info(
            f"Updating CodeEngine package '{name}' (id: {existing_pkg.id}) "
            f"because package already exists"
        )

        result = await existing_pkg.deploy_version(
            code=code,
            is_new_version=is_new_version,
            metadata=metadata,
            context=context,
            **context_kwargs,
        )

        # Refresh package state from API to reflect new version
        await logger.debug("Refreshing package state from API")
        return await cls.get_by_id(
            auth=auth,
            package_id=result["package_id"],
            context=context,
            **context_kwargs,
        )

    # 3) Otherwise, optionally create a brand new package
    if not create_if_missing:
        await logger.error(
            f"No existing CodeEngine package found for name '{name}', and create_if_missing=False"
        )
        raise CodeEngine_CRUD_Error(
            operation="upsert",
            message=f"No existing CodeEngine package found for name '{name}', and create_if_missing=False",
        )

    await logger.info(
        f"Creating CodeEngine package '{name}' because no existing package found"
    )

    return await cls.create(
        auth=auth,
        name=name,
        code=code,
        version=version,
        metadata=metadata,
        context=context,
        **context_kwargs,
    )

DomoCodeEngine_PackageVersion `dataclass` ¶

DomoCodeEngine_PackageVersion(
    parent: DomoEntity,
    package_id: str,
    version: str,
    language: str | None = None,
    description: str | None = None,
    createdby_id: int | None = None,
    released_dt: datetime | None = None,
    configuration: dict | None = None,
    createdby: DomoUser | None = None,
    accounts_mapping: list[int] | None = None,
    ml_model: list[str] | None = None,
    code: str | None = None,
    functions: dict | None = None,
    Manifest: CodeEngineManifest | None = None,
)

Bases: DomoSubEntity

CodeEngine Package Version subentity.

Represents a specific version of a CodeEngine package with its code, configuration, and function definitions.

Attributes:

Name	Type	Description
`parent`	`DomoEntity`	Reference to the parent DomoCodeEngine_Package (inherited from DomoSubEntity)
`auth`		Authentication object (inherited from parent via DomoSubEntity)
`package_id`	`str`	ID of the parent package
`version`	`str`	Version string (e.g., "1.0.0")
`language`	`str \| None`	Programming language
`description`	`str \| None`	Version description
`createdby_id`	`int \| None`	ID of user who created this version
`released_dt`	`datetime \| None`	Release datetime
`configuration`	`dict \| None`	Version configuration dict
`code`	`str \| None`	Source code string
`functions`	`dict \| None`	Function definitions dict
`Manifest`	`CodeEngineManifest \| None`	CodeEngineManifest for Python packages
`createdby`	`DomoUser \| None`	DomoUser who created this version
`accounts_mapping`	`list[int] \| None`	Account mapping configuration
`ml_model`	`list[str] \| None`	ML model configuration

delete `async` ¶

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

Delete this package version.

Parameters:

Name	Type	Description	Default
`return_raw`	`bool`	If True, return raw API response	`False`
`context`	`RouteContext \| None`	Optional RouteContext for API call configuration	`None`
`**context_kwargs`		Additional context parameters	`{}`

Returns:

Type	Description
	ResponseGetData containing deletion result

Raises:

Type	Description
`CodeEngine_CRUD_Error`	If version deletion fails

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

async def delete(
    self,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Delete this package version.

    Args:
        return_raw: If True, return raw API response
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        ResponseGetData containing deletion result

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

    await logger.info(
        f"Deleting version {self.version} from package {self.package_id}"
    )
    res = await codeengine_routes.delete_codeengine_package_version(
        auth=self.auth,
        package_id=self.package_id,
        version=self.version,
        context=context,
    )

    if return_raw:
        return res

    await logger.info(
        f"Successfully deleted version {self.version} from package {self.package_id}"
    )
    return res

download_source_code `async` ¶

download_source_code(
    download_folder: str = "./EXPORT/codeengine",
    file_name: str | None = None,
    replace_folder: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Download the source code for this version to a file.

Parameters:

Name	Type	Description	Default
`download_folder`	`str`	Folder to save the code	`'./EXPORT/codeengine'`
`file_name`	`str \| None`	Optional file name (auto-generated if not provided)	`None`
`replace_folder`	`bool`	Whether to replace existing folder	`False`
`context`	`RouteContext \| None`	Optional RouteContext for API call configuration	`None`
`**context_kwargs`		Additional context parameters (debug_api, session, etc.)	`{}`

Returns:

Type	Description
	Path to the downloaded file

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

async def download_source_code(
    self,
    download_folder: str = "./EXPORT/codeengine",
    file_name: str | None = None,
    replace_folder: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Download the source code for this version to a file.

    Args:
        download_folder: Folder to save the code
        file_name: Optional file name (auto-generated if not provided)
        replace_folder: Whether to replace existing folder
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters (debug_api, session, etc.)

    Returns:
        Path to the downloaded file
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    if not self.code:
        # Fetch code if not already loaded
        version_with_code = await self.get_by_id_and_version(
            auth=self.auth,
            package_id=self.package_id,
            version=self.version,
            params={"parts": "code"},
            context=context,
        )
        self.code = version_with_code.code

    extension = ".txt"
    if self.language == "PYTHON":
        extension = ".py"
    elif self.language == "JAVASCRIPT":
        extension = ".js"

    file_name = (
        file_name or f"{self.package_id}/{self.version}/functions{extension}"
    )

    file_path = os.path.join(download_folder, file_name)

    if self.Manifest:
        return self.Manifest.download_source_code(
            export_folder=os.path.join(download_folder, file_path),
            replace_folder=replace_folder,
        )

    dmuf.upsert_file(file_path, content=self.code, replace_folder=replace_folder)
    return file_path

export ¶

export(
    file_name: str | None = None,
    output_folder: str = "EXPORT/codeengine/",
)

Export the source code to a file.

Parameters:

Name	Type	Description	Default
`file_name`	`str \| None`	Optional file name (defaults to package_id)	`None`
`output_folder`	`str`	Output folder path	`'EXPORT/codeengine/'`

Returns:

Type	Description
	Path to the exported file

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

def export(
    self,
    file_name: str | None = None,
    output_folder: str = "EXPORT/codeengine/",
):
    """Export the source code to a file.

    Args:
        file_name: Optional file name (defaults to package_id)
        output_folder: Output folder path

    Returns:
        Path to the exported file
    """
    output_folder = (
        f"{output_folder}/" if not output_folder.endswith("/") else output_folder
    )

    dmuf.upsert_folder(output_folder)

    file_name = file_name or self.package_id
    language_key = (self.language or "").upper()
    extension = ExportExtension.PYTHON.value
    if language_key in ExportExtension.__members__:
        extension = ExportExtension[language_key].value

    file_name = dmuf.change_extension(file_name, extension)

    file_path = os.path.join(output_folder, file_name)

    with builtins.open(file_path, "w+", newline="\n", encoding="utf-8") as f:
        f.write(self.code)

    return file_path

from_dict `classmethod` ¶

from_dict(
    auth: DomoAuth,
    obj: dict[str, Any],
    package_id: str,
    language: str | None = None,
    is_supress_error: bool = True,
    parent: DomoCodeEngine_Package | None = None,
)

Create DomoCodeEngine_PackageVersion from API response.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`obj`	`dict[str, Any]`	API response dictionary	required
`package_id`	`str`	Parent package ID	required
`language`	`str \| None`	Programming language (optional, will use obj value if not provided)	`None`
`is_supress_error`	`bool`	Whether to suppress configuration errors	`True`
`parent`	`DomoCodeEngine_Package \| None`	Optional parent package reference (set after package creation if None)	`None`

Returns:

Type	Description
	DomoCodeEngine_PackageVersion instance

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

@classmethod
def from_dict(
    cls,
    auth: DomoAuth,
    obj: dict[str, Any],
    package_id: str,
    language: str | None = None,
    is_supress_error: bool = True,
    parent: DomoCodeEngine_Package | None = None,
):
    """Create DomoCodeEngine_PackageVersion from API response.

    Args:
        auth: Authentication object
        obj: API response dictionary
        package_id: Parent package ID
        language: Programming language (optional, will use obj value if not provided)
        is_supress_error: Whether to suppress configuration errors
        parent: Optional parent package reference (set after package creation if None)

    Returns:
        DomoCodeEngine_PackageVersion instance
    """
    language = (language or obj.get("language", "PYTHON")).upper()

    # DomoSubEntity requires parent, but we may not have it yet during package creation
    # Create a minimal parent object if needed (will be set properly after package creation)
    if parent is None:
        temp_parent = cls._create_temp_parent(auth=auth, package_id=package_id)
    else:
        temp_parent = parent

    # Only create Manifest if code is present and language is PYTHON
    code = obj.get("code")
    manifest = None
    if language == "PYTHON" and code:
        try:
            manifest = CodeEngineManifest.from_api(obj=obj)
        except (ValueError, AttributeError):
            # Code might be missing or invalid, skip manifest creation
            manifest = None

    domo_version = cls(
        parent=temp_parent,
        package_id=package_id,
        language=language,
        version=obj.get("version"),
        code=code,
        description=obj.get("description"),
        createdby_id=obj.get("createdBy"),
        released_dt=convert_string_to_datetime(obj.get("releasedOn")),
        configuration=obj.get("configuration"),
        Manifest=manifest,
        functions=obj.get("functions"),
    )

    # If we used a temp parent, update it to the real parent now
    if parent is not None and domo_version.parent is not parent:
        domo_version.parent = parent

    # Set configuration if available
    if domo_version.configuration:
        try:
            domo_version._set_configuration()
        except DomoCodeEngine_ConfigError:
            if not is_supress_error:
                raise

    return domo_version

get_by_id_and_version `async` `classmethod` ¶

get_by_id_and_version(
    auth: DomoAuth,
    package_id: str,
    version: str,
    language: str | None = None,
    params: dict | None = None,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    return_raw: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Retrieve a specific package version.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`package_id`	`str`	Package identifier	required
`version`	`str`	Version string	required
`language`	`str \| None`	Programming language (optional)	`None`
`params`	`dict \| None`	Optional query parameters	`None`
`debug_api`	`bool`	Enable API debugging	`False`
`debug_num_stacks_to_drop`	`int`	Stack frames to drop in logging	`2`
`return_raw`	`bool`	If True, return raw API response	`False`
`session`	`AsyncClient \| None`	Optional httpx session	`None`
`context`	`RouteContext \| None`	Optional RouteContext for API call configuration	`None`
`**context_kwargs`		Additional context parameters	`{}`

Returns:

Type	Description
	DomoCodeEngine_PackageVersion instance or ResponseGetData if return_raw=True

Raises:

Type	Description
`CodeEngine_GET_Error`	If version retrieval fails

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

@classmethod
async def get_by_id_and_version(
    cls,
    auth: DomoAuth,
    package_id: str,
    version: str,
    language: str | None = None,
    params: dict | None = None,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    return_raw: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Retrieve a specific package version.

    Args:
        auth: Authentication object
        package_id: Package identifier
        version: Version string
        language: Programming language (optional)
        params: Optional query parameters
        debug_api: Enable API debugging
        debug_num_stacks_to_drop: Stack frames to drop in logging
        return_raw: If True, return raw API response
        session: Optional httpx session
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        DomoCodeEngine_PackageVersion instance or ResponseGetData if return_raw=True

    Raises:
        CodeEngine_GET_Error: If version retrieval fails
    """
    params = params or {"parts": "functions,code"}

    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        **context_kwargs,
    )

    res = await codeengine_routes.get_codeengine_package_by_id_and_version(
        auth=auth,
        package_id=package_id,
        version=version,
        params=params,
        context=context,
    )

    if return_raw:
        return res

    # For get_by_id_and_version, we don't have the parent package yet
    # Create a temporary parent object with auth
    temp_parent = cls._create_temp_parent(auth=auth, package_id=package_id)

    return cls.from_dict(
        auth=auth,
        obj=res.response,
        package_id=package_id,
        language=language,
        parent=temp_parent,
    )

DomoCodeEngine_Packages `dataclass` ¶

DomoCodeEngine_Packages(auth: DomoAuth)

Bases: DomoManager

Manager class for CodeEngine package collections.

Provides methods to retrieve, search, and manage multiple CodeEngine packages.

Attributes:

Name	Type	Description
`auth`	`DomoAuth`	Authentication object

describe_all `async` ¶

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

Get descriptions of all CodeEngine packages and their functions.

Fetches each package's current version to extract function names and docstrings. Use this to quickly scan all deployed CodeEngine libraries and answer "is there already a function that does X?" before building a new one.

Returns:

Type	Description
`list[dict]`	List of dicts, each with name, description, language, current_version,
`list[dict]`	function_count, and functions list (name, description/docstring, inputs).

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

async def describe_all(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[dict]:
    """Get descriptions of all CodeEngine packages and their functions.

    Fetches each package's current version to extract function names and
    docstrings. Use this to quickly scan all deployed CodeEngine libraries
    and answer "is there already a function that does X?" before building
    a new one.

    Returns:
        List of dicts, each with name, description, language, current_version,
        function_count, and functions list (name, description/docstring, inputs).
    """
    packages = await self.get(context=context, **context_kwargs)
    return [await pkg.describe_async() for pkg in packages]

get `async` ¶

get(
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoCodeEngine_Package]

Get all CodeEngine packages.

Parameters:

Name	Type	Description	Default
`debug_api`	`bool`	Enable API debugging	`False`
`debug_num_stacks_to_drop`	`int`	Stack frames to drop in logging	`2`
`session`	`AsyncClient \| None`	Optional httpx session	`None`
`context`	`RouteContext \| None`	Optional RouteContext for API call configuration	`None`
`**context_kwargs`		Additional context parameters	`{}`

Returns:

Type	Description
`list[DomoCodeEngine_Package]`	List of DomoCodeEngine_Package instances

Raises:

Type	Description
`CodeEngine_GET_Error`	If package retrieval fails

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

async def get(
    self,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoCodeEngine_Package]:
    """Get all CodeEngine packages.

    Args:
        debug_api: Enable API debugging
        debug_num_stacks_to_drop: Stack frames to drop in logging
        session: Optional httpx session
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        List of DomoCodeEngine_Package instances

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

    res = await codeengine_routes.get_packages(
        auth=self.auth,
        context=context,
    )

    return [
        DomoCodeEngine_Package.from_dict(auth=self.auth, obj=obj)
        for obj in res.response
    ]

search `async` ¶

search(
    search_text: str | None = None,
    additional_filters_ls: list[dict] | None = None,
    maximum: int | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoCodeEngine_Package] | ResponseGetData

Search for CodeEngine packages via the datacenter search API.

Parameters:

Name	Type	Description	Default
`search_text`	`str \| None`	Optional text to filter packages by name	`None`
`additional_filters_ls`	`list[dict] \| None`	Additional filters to apply to the search	`None`
`maximum`	`int \| None`	Maximum number of results to return	`None`
`return_raw`	`bool`	If True, return raw API response	`False`
`context`	`RouteContext \| None`	Optional RouteContext for API call configuration	`None`
`**context_kwargs`		Additional context parameters	`{}`

Returns:

Type	Description
`list[DomoCodeEngine_Package] \| ResponseGetData`	List of DomoCodeEngine_Package instances, or ResponseGetData if return_raw=True

Raises:

Type	Description
`SearchDatacenterNoResultsFoundError`	If no packages match the search

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

async def search(
    self,
    search_text: str | None = None,
    additional_filters_ls: list[dict] | None = None,
    maximum: int | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoCodeEngine_Package] | rgd.ResponseGetData:
    """Search for CodeEngine packages via the datacenter search API.

    Args:
        search_text: Optional text to filter packages by name
        additional_filters_ls: Additional filters to apply to the search
        maximum: Maximum number of results to return
        return_raw: If True, return raw API response
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        List of DomoCodeEngine_Package instances, or ResponseGetData if return_raw=True

    Raises:
        SearchDatacenterNoResultsFoundError: If no packages match the search
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    await logger.debug("Searching CodeEngine packages via datacenter API")
    res = await datacenter_routes.search_datacenter(
        auth=self.auth,
        search_text=search_text,
        entity_type=datacenter_routes.Datacenter_Enum.PACKAGE.value,
        additional_filters_ls=additional_filters_ls,
        maximum=maximum,
        context=context,
    )

    if return_raw:
        return res

    packages = [
        DomoCodeEngine_Package.from_dict(auth=self.auth, obj=obj)
        for obj in res.response
    ]

    await logger.info(f"Found {len(packages)} CodeEngine packages via search")
    return packages

search_by_name `async` ¶

search_by_name(
    name: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoCodeEngine_Package]

Search for packages by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Package name to search for (case-insensitive partial match)	required
`context`	`RouteContext \| None`	Optional RouteContext for API call configuration	`None`
`**context_kwargs`		Additional context parameters (debug_api, session, debug_num_stacks_to_drop, etc.)	`{}`

Returns:

Type	Description
`list[DomoCodeEngine_Package]`	List of matching DomoCodeEngine_Package instances

Raises:

Type	Description
`SearchCodeEngine_NotFound`	If no packages match the search

Source code in src/crew_dcs/classes/DomoCodeEngine/CodeEngine.py

async def search_by_name(
    self,
    name: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoCodeEngine_Package]:
    """Search for packages by name.

    Args:
        name: Package name to search for (case-insensitive partial match)
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters (debug_api, session, debug_num_stacks_to_drop, etc.)

    Returns:
        List of matching DomoCodeEngine_Package instances

    Raises:
        SearchCodeEngine_NotFound: If no packages match the search
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    await logger.debug(f"Fetching all packages to search for name '{name}'")
    all_packages = await self.get(
        context=context,
    )
    await logger.debug(f"Found {len(all_packages)} total packages")

    matches = [
        pkg for pkg in all_packages if name.lower() in (pkg.name or "").lower()
    ]

    if not matches:
        await logger.warning(f"No packages found matching name '{name}'")
        raise SearchCodeEngineNotFoundError(
            search_criteria=f"name contains '{name}'",
        )

    await logger.info(f"Found {len(matches)} packages matching name '{name}'")
    return matches

ExportExtension ¶

Bases: Enum

File extension types for CodeEngine exports.

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,
    )

CodeEngine

CodeEngine ¶

Class Hierarchy¶

Relationships¶

Usage Example¶

CodeEngine_CRUD_Error ¶

CodeEngine_GET_Error ¶

DomoCodeEngine_ConfigError ¶

DomoCodeEngine_Package dataclass ¶

describe property ¶

display_url property ¶

associate_accounts async ¶

create async classmethod ¶

delete async ¶

deploy_release async classmethod ¶

deploy_version async ¶

describe_async async ¶

from_dict classmethod ¶

get_by_id async classmethod ¶

get_current_version async ¶

get_entity_by_id async classmethod ¶

get_owner async ¶

run_function async classmethod ¶

update_owner async ¶

upsert async classmethod ¶

DomoCodeEngine_PackageVersion dataclass ¶

delete async ¶

download_source_code async ¶

export ¶

from_dict classmethod ¶

get_by_id_and_version async classmethod ¶

DomoCodeEngine_Packages dataclass ¶

describe_all async ¶

get async ¶

search async ¶

search_by_name async ¶

ExportExtension ¶

SearchCodeEngineNotFoundError ¶

DomoCodeEngine_Package `dataclass` ¶

describe `property` ¶

display_url `property` ¶

associate_accounts `async` ¶

create `async` `classmethod` ¶

delete `async` ¶

deploy_release `async` `classmethod` ¶

deploy_version `async` ¶

describe_async `async` ¶

from_dict `classmethod` ¶

get_by_id `async` `classmethod` ¶

get_current_version `async` ¶

get_entity_by_id `async` `classmethod` ¶

get_owner `async` ¶

run_function `async` `classmethod` ¶

update_owner `async` ¶

upsert `async` `classmethod` ¶

DomoCodeEngine_PackageVersion `dataclass` ¶

delete `async` ¶

download_source_code `async` ¶

from_dict `classmethod` ¶

get_by_id_and_version `async` `classmethod` ¶

DomoCodeEngine_Packages `dataclass` ¶

describe_all `async` ¶

get `async` ¶

search `async` ¶

search_by_name `async` ¶