core

core

CodeEngine_Package_Parts

Bases: DomoEnumMixin, Enum

Enum for package parts parameter values.

delete_codeengine_package async

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

Delete a codeengine package by ID.

Parameters:

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

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing deletion result

Raises:

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

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

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

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

    Returns:
        ResponseGetData object containing deletion result

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

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

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

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

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

    if return_raw:
        return res

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

    return res

delete_codeengine_package_version async

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

Delete a specific version of a codeengine package.

Parameters:

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

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing deletion result

Raises:

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

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

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

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

    Returns:
        ResponseGetData object containing deletion result

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

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

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

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

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

    if return_raw:
        return res

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

    return res

execute_codeengine_function async

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

Execute one function of a deployed package version.

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

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

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

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

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

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

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

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

    if return_raw:
        return res

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

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

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

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

    metadata = res.response

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

    return res

get_codeengine_package_by_id async

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

Retrieve a specific codeengine package by ID.

Parameters:

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

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing package data

Raises:

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

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

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

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

    Returns:
        ResponseGetData object containing package data

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

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

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

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

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

    if return_raw:
        return res

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

    return res

get_codeengine_package_by_id_and_version async

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

Retrieve a specific codeengine package by ID and version.

Parameters:

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

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing package version data

Raises:

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

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

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

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

    Returns:
        ResponseGetData object containing package version data

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

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

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

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

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

    if return_raw:
        return res

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

    return res

get_current_package_version async

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

Get the current version of a CodeEngine package.

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

Parameters:

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

Returns:

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

Raises:

Type	Description
CodeEngine_GET_Error	If package retrieval fails or version not found

Example

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

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

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

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

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

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

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

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

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

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

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

    return current_version

get_package_versions async

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

Retrieve all versions of a codeengine package.

Each package can have one or many versions.

Parameters:

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

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing package versions

Raises:

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

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

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

    Each package can have one or many versions.

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

    Returns:
        ResponseGetData object containing package versions

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

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

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

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

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

    if return_raw:
        return res

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

    return res

get_packages async

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

Retrieve all codeengine packages.

Parameters:

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

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing package list

Raises:

Type	Description
CodeEngine_GET_Error	If package retrieval fails

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

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

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

    Returns:
        ResponseGetData object containing package list

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

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

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

    if return_raw:
        return res

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

    return res

test_package_is_identical async

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

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

Parameters:

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

Returns:

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

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

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

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

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

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

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

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

test_package_is_released async

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

Test if a package version is already released.

Parameters:

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

Returns:

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

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

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

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

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

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

    return existing_package.get("released", False)

update_codeengine_package_owner async

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

Update the owner of a codeengine package.

Parameters:

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

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing updated package data

Raises:

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

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

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

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

    Returns:
        ResponseGetData object containing updated package data

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

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

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

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

    body = {"owner": owner_id}

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

    if return_raw:
        return res

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

    return res