sandbox

sandbox

Sandbox_CRUD_Error

Sandbox_CRUD_Error(
    operation: str,
    repository_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs
)

Bases: RouteError

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

Source code in src/crew_dcs/routes/sandbox.py

def __init__(
    self,
    operation: str,
    repository_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs,
):
    super().__init__(
        message=message or f"Sandbox {operation} operation failed",
        entity_id=repository_id,
        res=res,
        **kwargs,
    )

Sandbox_GET_Error

Sandbox_GET_Error(
    repository_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs
)

Bases: RouteError

Raised when sandbox retrieval operations fail.

Source code in src/crew_dcs/routes/sandbox.py

def __init__(
    self,
    repository_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs,
):
    super().__init__(
        message=message or "Sandbox retrieval failed",
        entity_id=repository_id,
        res=res,
        **kwargs,
    )

create_commit async

create_commit(
    auth: DomoAuth,
    repository_id: str,
    commit: CommitCreate | dict[str, Any],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Create a commit (POST .../repositories/{id}/commitRequests).

commit may be a CommitCreate (adds a fresh pusherEventId) or a raw dict, sent as-is.

Source code in src/crew_dcs/routes/sandbox.py

@gd.route_function
@log_call(level_name="route", config=_ROUTE_LOG_CONFIG)
async def create_commit(
    auth: DomoAuth,
    repository_id: str,
    commit: CommitCreate | dict[str, Any],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create a commit (POST .../repositories/{id}/commitRequests).

    ``commit`` may be a ``CommitCreate`` (adds a fresh ``pusherEventId``) or a
    raw dict, sent as-is.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)
    return await _post_repo(
        auth,
        repository_id,
        "commitRequests",
        _to_body(CommitCreate, commit),
        "create commit",
        return_raw,
        context,
    )

get_commit_requests async

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

Get a repository's commit requests (GET .../repositories/{id}/commitRequests).

Source code in src/crew_dcs/routes/sandbox.py

@gd.route_function
@log_call(level_name="route", config=_ROUTE_LOG_CONFIG)
async def get_commit_requests(
    auth: DomoAuth,
    repository_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get a repository's commit requests (GET .../repositories/{id}/commitRequests)."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    return await _get_repo_subresource(
        auth, repository_id, "commitRequests", return_raw, context
    )

get_commits async

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

Get a repository's commits (GET .../repositories/{id}/commits).

Source code in src/crew_dcs/routes/sandbox.py

@gd.route_function
@log_call(level_name="route", config=_ROUTE_LOG_CONFIG)
async def get_commits(
    auth: DomoAuth,
    repository_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get a repository's commits (GET .../repositories/{id}/commits)."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    return await _get_repo_subresource(
        auth, repository_id, "commits", return_raw, context
    )

get_instance_access async

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

Get instances with access to a repository (GET .../repositories/{id}/access).

Source code in src/crew_dcs/routes/sandbox.py

@gd.route_function
@log_call(level_name="route", config=_ROUTE_LOG_CONFIG)
async def get_instance_access(
    auth: DomoAuth,
    repository_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get instances with access to a repository (GET .../repositories/{id}/access)."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    return await _get_repo_subresource(
        auth, repository_id, "access", return_raw, context
    )

get_instances async

get_instances(
    auth: DomoAuth,
    limit: int = 250,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

List authorized instances (GET /version/v1/authorizations).

Raises:

Type	Description
Sandbox_GET_Error	If retrieval fails

Source code in src/crew_dcs/routes/sandbox.py

@gd.route_function
@log_call(level_name="route", config=_ROUTE_LOG_CONFIG)
async def get_instances(
    auth: DomoAuth,
    limit: int = 250,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """List authorized instances (GET /version/v1/authorizations).

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

    res = await gd.get_data(
        auth=auth,
        method="GET",
        url=f"{_base_url(auth)}/authorizations",
        params={"limit": limit},
        context=context,
    )

    if return_raw:
        return res

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

    return res

get_repo_from_id async

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

Get a sandbox repository by ID.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object	required
repository_id	str	Repository identifier	required
return_raw	bool	Return raw response without processing	False
context	RouteContext | None	Optional RouteContext for request configuration	None

Returns:

Type	Description
ResponseGetData	ResponseGetData object

Raises:

Type	Description
Sandbox_GET_Error	If retrieval fails

Source code in src/crew_dcs/routes/sandbox.py

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

    Args:
        auth: Authentication object
        repository_id: Repository identifier
        return_raw: Return raw response without processing
        context: Optional RouteContext for request configuration

    Returns:
        ResponseGetData object

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

    url = f"https://{auth.domo_instance}.domo.com/api/version/v1/repositories/{repository_id}"

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

    if return_raw:
        return res

    if not res.is_success:
        raise Sandbox_GET_Error(repository_id=repository_id, res=res)

    return res

get_repo_permissions async

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

Get a repository's user/group permissions (GET .../repositories/{id}/permissions).

Source code in src/crew_dcs/routes/sandbox.py

@gd.route_function
@log_call(level_name="route", config=_ROUTE_LOG_CONFIG)
async def get_repo_permissions(
    auth: DomoAuth,
    repository_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get a repository's user/group permissions (GET .../repositories/{id}/permissions)."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    return await _get_repo_subresource(
        auth, repository_id, "permissions", return_raw, context
    )

promote_and_seed async

promote_and_seed(
    auth: DomoAuth,
    repository_id: str,
    deployment_id: str,
    request: PromoteAndSeedRequest | dict[str, Any],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Promote and link/seed a commit (POST .../deployments/{deploymentId}/promoteAndSeed).

request may be a PromoteAndSeedRequest (adds a fresh pusherEventId) or a raw dict, sent as-is.

Source code in src/crew_dcs/routes/sandbox.py

@gd.route_function
@log_call(level_name="route", config=_ROUTE_LOG_CONFIG)
async def promote_and_seed(
    auth: DomoAuth,
    repository_id: str,
    deployment_id: str,
    request: PromoteAndSeedRequest | dict[str, Any],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Promote and link/seed a commit (POST .../deployments/{deploymentId}/promoteAndSeed).

    ``request`` may be a ``PromoteAndSeedRequest`` (adds a fresh
    ``pusherEventId``) or a raw dict, sent as-is.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)
    return await _post_repo(
        auth,
        repository_id,
        f"deployments/{deployment_id}/promoteAndSeed",
        _to_body(PromoteAndSeedRequest, request),
        "promote and seed",
        return_raw,
        context,
    )

search_commit_history async

search_commit_history(
    auth: DomoAuth,
    filters: dict | None = None,
    sort: str = "completed",
    order: str = "desc",
    date_filters: dict | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Search commit history (POST /version/v1/commitRequests/search), paginated.

Request shape follows the Postman example; the response list key is UNVERIFIED (tries commitRequests, commits, results, items, or a bare list) -- rests on the endpoint shape, not on a runnable proof.

Parameters:

Name	Type	Description	Default
filters	dict | None	e.g. {"repositoryName": [], "commitName": [], "status": []}	None

Source code in src/crew_dcs/routes/sandbox.py

@gd.route_function
@log_call(level_name="route", config=_ROUTE_LOG_CONFIG)
async def search_commit_history(
    auth: DomoAuth,
    filters: dict | None = None,
    sort: str = "completed",
    order: str = "desc",
    date_filters: dict | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Search commit history (POST /version/v1/commitRequests/search), paginated.

    Request shape follows the Postman example; the response list key is
    UNVERIFIED (tries ``commitRequests``, ``commits``, ``results``, ``items``,
    or a bare list) -- rests on the endpoint shape, not on a runnable proof.

    Args:
        filters: e.g. ``{"repositoryName": [], "commitName": [], "status": []}``
    """
    context = RouteContext.build_context(context=context, **context_kwargs)
    return await _search_history(
        auth,
        "commitRequests/search",
        _COMMIT_SEARCH_KEYS,
        _search_body(filters, sort, order, date_filters),
        return_raw,
        context,
    )

search_promotion_history async

search_promotion_history(
    auth: DomoAuth,
    filters: dict | None = None,
    sort: str = "completed",
    order: str = "desc",
    date_filters: dict | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Search promotion history (POST /version/v1/promotions/search), paginated.

Request shape follows the Postman example; the response list key is UNVERIFIED (tries promotions, results, items, or a bare list) -- rests on the endpoint shape, not on a runnable proof.

Parameters:

Name	Type	Description	Default
filters	dict | None	e.g. {"repositoryName": [], "commitName": [], "status": [], "userId": []}	None

Source code in src/crew_dcs/routes/sandbox.py

@gd.route_function
@log_call(level_name="route", config=_ROUTE_LOG_CONFIG)
async def search_promotion_history(
    auth: DomoAuth,
    filters: dict | None = None,
    sort: str = "completed",
    order: str = "desc",
    date_filters: dict | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Search promotion history (POST /version/v1/promotions/search), paginated.

    Request shape follows the Postman example; the response list key is
    UNVERIFIED (tries ``promotions``, ``results``, ``items``, or a bare list)
    -- rests on the endpoint shape, not on a runnable proof.

    Args:
        filters: e.g. ``{"repositoryName": [], "commitName": [], "status": [], "userId": []}``
    """
    context = RouteContext.build_context(context=context, **context_kwargs)
    return await _search_history(
        auth,
        "promotions/search",
        _PROMOTION_SEARCH_KEYS,
        _search_body(filters, sort, order, date_filters),
        return_raw,
        context,
    )

toggle_allow_same_instance_promotion async

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

Toggle the allow same instance promotion setting.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object	required
is_enabled	bool	Whether to enable same instance promotion	required
return_raw	bool	Return raw response without processing	False
context	RouteContext | None	Optional RouteContext for request configuration	None

Returns:

Type	Description
ResponseGetData	ResponseGetData object

Raises:

Type	Description
Sandbox_CRUD_Error	If the operation fails

Source code in src/crew_dcs/routes/sandbox.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def toggle_allow_same_instance_promotion(
    auth: DomoAuth,
    is_enabled: bool,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Toggle the allow same instance promotion setting.

    Args:
        auth: Authentication object
        is_enabled: Whether to enable same instance promotion
        return_raw: Return raw response without processing
        context: Optional RouteContext for request configuration

    Returns:
        ResponseGetData object

    Raises:
        Sandbox_CRUD_Error: If the operation fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/version/v1/settings"

    body = {"allowSelfPromotion": is_enabled}

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

    if return_raw:
        return res

    if not res.is_success:
        raise Sandbox_CRUD_Error(operation="toggle same instance promotion", res=res)

    return res

update_instance_alias async

update_instance_alias(
    auth: DomoAuth,
    alias_update: InstanceAliasUpdate | dict[str, Any],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Set an instance alias (POST /version/v1/authorizations/aliases).

Parameters:

Name	Type	Description	Default
alias_update	InstanceAliasUpdate | dict[str, Any]	InstanceAliasUpdate or a raw dict {"aliasedDomain": ..., "alias": ...}	required

Raises:

Type	Description
Sandbox_CRUD_Error	If the operation fails

Source code in src/crew_dcs/routes/sandbox.py

@gd.route_function
@log_call(level_name="route", config=_ROUTE_LOG_CONFIG)
async def update_instance_alias(
    auth: DomoAuth,
    alias_update: InstanceAliasUpdate | dict[str, Any],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Set an instance alias (POST /version/v1/authorizations/aliases).

    Args:
        alias_update: ``InstanceAliasUpdate`` or a raw dict
            ``{"aliasedDomain": ..., "alias": ...}``

    Raises:
        Sandbox_CRUD_Error: If the operation fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await gd.get_data(
        auth=auth,
        method="POST",
        url=f"{_base_url(auth)}/authorizations/aliases",
        body=_to_body(InstanceAliasUpdate, alias_update),
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success:
        raise Sandbox_CRUD_Error(operation="update instance alias", res=res)

    return res

update_repo_instance_access async

update_repo_instance_access(
    auth: DomoAuth,
    repository_id: str,
    domains: list[str],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Set the instances with access to a repository (POST .../repositories/{id}/access).

Body is a bare list of domain strings. The Postman collection lists the path as /repositories{repositoryId}/access (missing slash); the slash is assumed a typo and this rests on the GET sibling's path, not a runnable proof.

Source code in src/crew_dcs/routes/sandbox.py

@gd.route_function
@log_call(level_name="route", config=_ROUTE_LOG_CONFIG)
async def update_repo_instance_access(
    auth: DomoAuth,
    repository_id: str,
    domains: list[str],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Set the instances with access to a repository (POST .../repositories/{id}/access).

    Body is a bare list of domain strings. The Postman collection lists the path
    as ``/repositories{repositoryId}/access`` (missing slash); the slash is
    assumed a typo and this rests on the GET sibling's path, not a runnable proof.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)
    return await _post_repo(
        auth,
        repository_id,
        "access",
        list(domains),
        "update instance access",
        return_raw,
        context,
    )

update_repo_permissions async

update_repo_permissions(
    auth: DomoAuth,
    repository_id: str,
    updates: list[
        RepositoryPermissionUpdate | dict[str, Any]
    ],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Update repository user/group permissions (POST .../repositories/{id}/permissions).

Each item may be a RepositoryPermissionUpdate or a raw dict; the unused userId/groupId is sent as '' per the Postman example.

Source code in src/crew_dcs/routes/sandbox.py

@gd.route_function
@log_call(level_name="route", config=_ROUTE_LOG_CONFIG)
async def update_repo_permissions(
    auth: DomoAuth,
    repository_id: str,
    updates: list[RepositoryPermissionUpdate | dict[str, Any]],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Update repository user/group permissions (POST .../repositories/{id}/permissions).

    Each item may be a ``RepositoryPermissionUpdate`` or a raw dict; the unused
    ``userId``/``groupId`` is sent as ``''`` per the Postman example.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)
    body = {
        "repositoryPermissionUpdates": [
            _to_body(RepositoryPermissionUpdate, u) for u in updates
        ]
    }
    return await _post_repo(
        auth,
        repository_id,
        "permissions",
        body,
        "update permissions",
        return_raw,
        context,
    )