ownership

ownership

get_owned_alerts async

get_owned_alerts(
    auth: DomoAuth,
    user_id: int | str,
    limit: int = 50,
    offset: int = 0,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

List alerts owned by a specific user.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for API requests	required
user_id	int | str	ID of the owning user	required
limit	int	Maximum number of alerts to return	50
offset	int	Pagination offset	0
return_raw	bool	Return raw response without processing	False
context	RouteContext | None	RouteContext for request configuration	None

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing the user's alerts

Raises:

Type	Description
Alert_GET_Error	If alert listing fails

Source code in src/crew_dcs/routes/alerts/ownership.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def get_owned_alerts(
    auth: DomoAuth,
    user_id: int | str,
    limit: int = 50,
    offset: int = 0,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """List alerts owned by a specific user.

    Args:
        auth: Authentication object for API requests
        user_id: ID of the owning user
        limit: Maximum number of alerts to return
        offset: Pagination offset
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object containing the user's alerts

    Raises:
        Alert_GET_Error: If alert listing fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/social/v4/alerts"
    params = {"ownerId": user_id, "limit": limit, "offset": offset}

    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 Alert_GET_Error(res=res)
    return res

transfer_alerts async

transfer_alerts(
    auth: DomoAuth,
    from_user_id: int | str,
    to_user_id: int | str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[ResponseGetData]

Transfer all alerts from one user to another.

Fetches all alerts owned by from_user_id, then updates each one's owner to to_user_id.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for API requests	required
from_user_id	int | str	ID of the current owner	required
to_user_id	int | str	ID of the new owner	required
context	RouteContext | None	RouteContext for request configuration	None

Returns:

Type	Description
list[ResponseGetData]	List of per-alert update ResponseGetData objects

Raises:

Type	Description
Alert_GET_Error	If listing the owned alerts fails
Alert_CRUD_Error	If any owner update fails

Source code in src/crew_dcs/routes/alerts/ownership.py

async def transfer_alerts(
    auth: DomoAuth,
    from_user_id: int | str,
    to_user_id: int | str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[rgd.ResponseGetData]:
    """Transfer all alerts from one user to another.

    Fetches all alerts owned by from_user_id, then updates each one's owner
    to to_user_id.

    Args:
        auth: Authentication object for API requests
        from_user_id: ID of the current owner
        to_user_id: ID of the new owner
        context: RouteContext for request configuration

    Returns:
        List of per-alert update ResponseGetData objects

    Raises:
        Alert_GET_Error: If listing the owned alerts fails
        Alert_CRUD_Error: If any owner update fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    owned_res = await get_owned_alerts(auth=auth, user_id=from_user_id, context=context)

    alerts = owned_res.response or []

    results = []
    for alert in alerts:
        res = await update_alert_owner(
            auth=auth,
            alert_id=str(alert["id"]),
            to_user_id=to_user_id,
            context=context,
        )
        results.append(res)

    return results

update_alert_owner async

update_alert_owner(
    auth: DomoAuth,
    alert_id: str,
    to_user_id: int | str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Change the owner of a single alert.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for API requests	required
alert_id	str	ID of the alert to update	required
to_user_id	int | str	ID of the new owner	required
return_raw	bool	Return raw response without processing	False
context	RouteContext | None	RouteContext for request configuration	None

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing the updated alert

Raises:

Type	Description
Alert_CRUD_Error	If the owner update fails

Source code in src/crew_dcs/routes/alerts/ownership.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def update_alert_owner(
    auth: DomoAuth,
    alert_id: str,
    to_user_id: int | str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Change the owner of a single alert.

    Args:
        auth: Authentication object for API requests
        alert_id: ID of the alert to update
        to_user_id: ID of the new owner
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object containing the updated alert

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

    url = f"https://{auth.domo_instance}.domo.com/api/social/v4/alerts/{alert_id}"
    body = {"id": int(alert_id), "owner": int(to_user_id)}

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

    if return_raw:
        return res
    if not res.is_success:
        raise Alert_CRUD_Error(
            operation="update owner", alert_id=str(alert_id), res=res
        )
    return res