Skip to content

DomoCodeEngine

DomoCodeEngine

DomoCodeEngine module for managing CodeEngine packages and versions.

This module provides classes for interacting with Domo CodeEngine including: - Package management (DomoCodeEngine_Package, DomoCodeEngine_Packages) - Version control (DomoCodeEngine_PackageVersion) - Manifest handling (CodeEngineManifest, CodeEngineManifest_Function)

CodeEngineManifest dataclass 

CodeEngineManifest(
    functions: list[CodeEngineManifest_Function],
    configuration: dict[str, Any],
    raw: str = None,
)

from_python_file classmethod 

from_python_file(
    file_path: str, accounts_mapping: list[Any]
) -> CodeEngineManifest

Analyze all functions in a Python script file and return a structured manifest.

This is the main public method that orchestrates the complete analysis process. It loads the specified file, parses its AST, and generates comprehensive metadata descriptions for every function found in the script, returning them as a Manifest object.

Parameters:

Name Type Description Default
file_path str

Path to the Python script file to analyze

 required

Returns:

Name Type Description
Manifest CodeEngineManifest

A manifest object containing: - functions (list[FunctionMetadata]): list of function metadata objects - configuration (dict[str, Any]): Configuration dictionary with accountsMapping
Error Handling
  • Returns a Manifest with empty functions list if file loading fails
  • Prints error messages for individual function analysis failures but continues processing
  • Skips functions that cannot be analyzed due to errors
Example

analyzer = CodeEngineScriptAnalyzer() manifest = analyzer.analyze_all_functions('my_script.py') for func in manifest.functions: ... print(f"Function: {func.name} ({len(func.inputs)} parameters)")

Note

This method walks the entire AST and processes all function definitions found, including nested functions and methods within classes.

Source code in src/crew_dcs/classes/DomoCodeEngine/Manifest.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@classmethod
def from_python_file(
    cls, file_path: str, accounts_mapping: list[Any]
) -> "CodeEngineManifest":
    """
    Analyze all functions in a Python script file and return a structured manifest.

    This is the main public method that orchestrates the complete analysis process.
    It loads the specified file, parses its AST, and generates comprehensive metadata
    descriptions for every function found in the script, returning them as a Manifest object.

    Args:
        file_path (str): Path to the Python script file to analyze

    Returns:
        Manifest: A manifest object containing:
            - functions (list[FunctionMetadata]): list of function metadata objects
            - configuration (dict[str, Any]): Configuration dictionary with accountsMapping

    Error Handling:
        - Returns a Manifest with empty functions list if file loading fails
        - Prints error messages for individual function analysis failures but continues processing
        - Skips functions that cannot be analyzed due to errors

    Example:
        >>> analyzer = CodeEngineScriptAnalyzer()
        >>> manifest = analyzer.analyze_all_functions('my_script.py')
        >>> for func in manifest.functions:
        ...     print(f"Function: {func.name} ({len(func.inputs)} parameters)")

    Note:
        This method walks the entire AST and processes all function definitions found,
        including nested functions and methods within classes.
    """
    content = dmcv.convert_python_to_ast_module(
        python_file_path=file_path, return_str=True
    )

    ast_module = dmcv.convert_python_to_ast(python_str=content, return_str=False)

    return cls(
        functions=[
            CodeEngineManifest_Function.from_ast_function(ast_fn, content=content)
            for ast_fn in dmcv.extract_ast_functions(ast_module)
        ],
        configuration={"accountsMapping": accounts_mapping or []},
    )

from_python_string classmethod 

from_python_string(
    python_str: str,
    accounts_mapping: list[Any],
    validate: bool = True,
) -> CodeEngineManifest

Analyze all functions in a Python script string and return a structured manifest.

This is the main public method that orchestrates the complete analysis process. It loads the specified string, parses its AST, and generates comprehensive metadata descriptions for every function found in the script, returning them as a Manifest object.

Parameters:

Name Type Description Default
python_str str

Python script content to analyze

 required

Returns:

Name Type Description
Manifest CodeEngineManifest

A manifest object containing: - functions (list[FunctionMetadata]): list of function metadata objects - configuration (dict[str, Any]): Configuration dictionary with accountsMapping
Error Handling
  • Returns a Manifest with empty functions list if string loading fails
  • Prints error messages for individual function analysis failures but continues processing
  • Skips functions that cannot be analyzed due to errors
Example

analyzer = CodeEngineScriptAnalyzer() manifest = analyzer.analyze_all_functions('def my_func(x): return x') for func in manifest.functions: ... print(f"Function: {func.name} ({len(func.inputs)} parameters)")

Note

This method walks the entire AST and processes all function definitions found, including nested functions and methods within classes.

Source code in src/crew_dcs/classes/DomoCodeEngine/Manifest.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
@classmethod
def from_python_string(
    cls, python_str: str, accounts_mapping: list[Any], validate: bool = True
) -> "CodeEngineManifest":
    """
    Analyze all functions in a Python script string and return a structured manifest.

    This is the main public method that orchestrates the complete analysis process.
    It loads the specified string, parses its AST, and generates comprehensive metadata
    descriptions for every function found in the script, returning them as a Manifest object.

    Args:
        python_str (str): Python script content to analyze

    Returns:
        Manifest: A manifest object containing:
            - functions (list[FunctionMetadata]): list of function metadata objects
            - configuration (dict[str, Any]): Configuration dictionary with accountsMapping

    Error Handling:
        - Returns a Manifest with empty functions list if string loading fails
        - Prints error messages for individual function analysis failures but continues processing
        - Skips functions that cannot be analyzed due to errors

    Example:
        >>> analyzer = CodeEngineScriptAnalyzer()
        >>> manifest = analyzer.analyze_all_functions('def my_func(x): return x')
        >>> for func in manifest.functions:
        ...     print(f"Function: {func.name} ({len(func.inputs)} parameters)")

    Note:
        This method walks the entire AST and processes all function definitions found,
        including nested functions and methods within classes.
    """
    ast_module = dmcv.convert_python_to_ast_module(
        python_str=python_str, return_str=False
    )

    manifest = cls(
        functions=[
            CodeEngineManifest_Function.from_ast_function_def(
                ast_fn, original_module_string=python_str
            )
            for ast_fn in dmcv.extract_ast_functions(ast_module)
        ],
        configuration={"accountsMapping": accounts_mapping or []},
    )
    # validate=False on the READ path: the description cap is a write-side
    # constraint, and refusing to parse a package that already exists in
    # Domo would break inspection of it for no benefit.
    if validate:
        manifest.validate_descriptions()
    return manifest

to_dict 

to_dict() -> dict[str, Any]

Convert manifest to dictionary format for API submission.

Returns:

Type Description
dict[str, Any]

Dictionary with 'functions' and 'configuration' keys.
Source code in src/crew_dcs/classes/DomoCodeEngine/Manifest.py 
180
181
182
183
184
185
186
187
188
189
def to_dict(self) -> dict[str, Any]:
    """Convert manifest to dictionary format for API submission.

    Returns:
        Dictionary with 'functions' and 'configuration' keys.
    """
    return {
        "functions": [fn.to_dict() for fn in self.functions],
        "configuration": self.configuration,
    }

validate_descriptions 

validate_descriptions() -> None

Raise if any function docstring exceeds Domo's description cap.

Called before the manifest is sent so an over-long docstring fails here, naming the function and the overage, instead of as Domo's anonymous 400. See MAX_FUNCTION_DESCRIPTION_LEN for how the limit was established.

Source code in src/crew_dcs/classes/DomoCodeEngine/Manifest.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def validate_descriptions(self) -> None:
    """Raise if any function docstring exceeds Domo's description cap.

    Called before the manifest is sent so an over-long docstring fails here,
    naming the function and the overage, instead of as Domo's anonymous 400.
    See MAX_FUNCTION_DESCRIPTION_LEN for how the limit was established.
    """
    too_long = [
        (fn.name, len(fn.description or ""))
        for fn in self.functions
        if len(fn.description or "") > MAX_FUNCTION_DESCRIPTION_LEN
    ]
    if not too_long:
        return

    detail = "; ".join(
        f"{name} is {size} chars ({size - MAX_FUNCTION_DESCRIPTION_LEN} over)"
        for name, size in too_long
    )
    raise CodeEngine_InvalidPackageError(
        message=(
            f"Function docstring exceeds Domo's {MAX_FUNCTION_DESCRIPTION_LEN}-character "
            f"description limit: {detail}. Domo rejects this with a 400 that names no "
            f"field. Move the extra prose into module-level comments, which are not capped."
        )
    )

CodeEngineManifest_Argument dataclass 

CodeEngineManifest_Argument(
    annotation_text: str,
    has_default_value: bool = None,
    ast_arg: arg = None,
    schema_type: PythonTypeToSchemaType = None,
    is_list: bool = False,
    is_nullable: bool = False,
    name: str = None,
    default_value: str = None,
    display_name: str = None,
    value: str = None,
    children: list[Any] = list(),
)

Bases: CodeEngine_Argument

display class for formatting AST Argument

CodeEngineManifest_Function dataclass 

CodeEngineManifest_Function(
    ast_fn: FunctionDef,
    name: str,
    display_name: str,
    description: str,
    is_private: bool,
    editor_start_index: int,
    has_return: bool,
    output: CodeEngineManifest_Argument,
    editor_end_index: int = None,
    content: str = None,
    input_args: list[CodeEngineManifest_Argument] = list(),
    _target_from_api: dict = None,
)

from_ast_function_def classmethod 

from_ast_function_def(
    ast_fn: FunctionDef, original_module_string
) -> CodeEngineManifest_Function

Create a complete function description as a CodeEngineManifest_Function object.

Combines all extracted function metadata into a structured FunctionMetadata dataclass instance that includes the function name, parameters, return type, docstring, and editor positioning information required for code engine integration.

Parameters:

Name Type Description Default
function_node FunctionDef

The AST node representing the function to describe

 required

Returns:

Name Type Description
FunctionMetadata CodeEngineManifest_Function

Complete function metadata object containing: - name (str): Original function name - displayName (str): Human-readable function name - description (str): Function docstring - inputs (list[Dict]): Parameter descriptions - output (Dict): Return value description - variables (Dict): Variable descriptions (copy of inputs) - isPrivate (bool): Whether function is private (based on underscore prefix) - editorStartIndex (int): Character position in source file - hasReturn (bool): Whether function has return type annotation
Note

The isPrivate field is determined by checking if the function name starts with an underscore, following Python naming conventions.

Source code in src/crew_dcs/classes/DomoCodeEngine/Manifest_Function.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
@classmethod
def from_ast_function_def(
    cls, ast_fn: ast.FunctionDef, original_module_string
) -> "CodeEngineManifest_Function":
    """
    Create a complete function description as a CodeEngineManifest_Function object.

    Combines all extracted function metadata into a structured FunctionMetadata
    dataclass instance that includes the function name, parameters, return type,
    docstring, and editor positioning information required for code engine integration.

    Args:
        function_node (ast.FunctionDef): The AST node representing the function to describe

    Returns:
        FunctionMetadata: Complete function metadata object containing:
            - name (str): Original function name
            - displayName (str): Human-readable function name
            - description (str): Function docstring
            - inputs (list[Dict]): Parameter descriptions
            - output (Dict): Return value description
            - variables (Dict): Variable descriptions (copy of inputs)
            - isPrivate (bool): Whether function is private (based on underscore prefix)
            - editorStartIndex (int): Character position in source file
            - hasReturn (bool): Whether function has return type annotation

    Note:
        The isPrivate field is determined by checking if the function name starts
        with an underscore, following Python naming conventions.
    """
    function_name = ast_fn.name
    is_private = function_name.startswith("_")

    input_args = [
        CodeEngineManifest_Argument.from_ast_arg(arg, ast_fn)
        for arg in ast_fn.args.args
    ]

    output_arg = CodeEngineManifest_Argument.from_ast_function_return_arg(ast_fn)

    editor_start_index = calculate_character_index_of_line_number(
        ast_fn.lineno - 1, full_text=original_module_string
    )

    editor_end_index = calculate_character_index_of_line_number(
        ast_fn.end_lineno - 1, full_text=original_module_string
    )

    content = "\n".join(
        original_module_string.split("\n")[
            ast_fn.lineno - 1 : ast_fn.end_lineno - 1
        ]
        if ast_fn.end_lineno
        else []
    )
    # content = original_module_string[editor_start_index:editor_end_index] if editor_end_index else None

    return cls(
        ast_fn=ast_fn,
        name=function_name,
        is_private=is_private,
        display_name=dmcv.convert_programming_text_to_title_case(function_name),
        description=extract_function_docstring(ast_fn),
        input_args=input_args,
        output=output_arg,
        editor_start_index=editor_start_index,
        editor_end_index=editor_end_index,
        content=content,
        has_return=ast_fn.returns is not None,
    )

to_dict 

to_dict() -> dict

Serialize the function metadata and arguments to a manifest-compatible dictionary.

Source code in src/crew_dcs/classes/DomoCodeEngine/Manifest_Function.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def to_dict(self) -> dict:
    """
    Serialize the function metadata and arguments to a manifest-compatible dictionary.
    """
    return {
        "name": self.name,
        "displayName": self.display_name,
        "description": self.description,
        "isPrivate": self.is_private,
        "inputs": [
            arg.to_dict(include_orig=False) if hasattr(arg, "to_dict") else arg
            for arg in self.input_args
        ],
        "code": self.content,
        "output": (
            self.output.to_dict(include_orig=False)
            if hasattr(self.output, "to_dict")
            else self.output
        ),
    }

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 
60
61
62
63
64
65
66
67
68
69
70
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 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
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 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
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 
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
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 
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
@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 
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
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 
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
@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 
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
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 
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
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 
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
@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 
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
@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 
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
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 
783
784
785
786
787
788
789
@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 
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
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 
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
@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 
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
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 
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
@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 
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
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 
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
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 
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
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 
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
@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 
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
@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 
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
1728
1729
1730
1731
1732
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 
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
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 
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
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 
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
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 
43
44
45
46
47
48
49
50
51
52
53
54
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,
    )

Modules