DomoAlert

DomoAlert

DomoAlert dataclass

DomoAlert(
    auth: DomoAuth,
    id: str,
    raw: dict,
    name: str | None = None,
    owner_id: str | None = None,
    enabled: bool | None = None,
)

Bases: DomoEntity

Represents a Domo Alert.

create async classmethod

create(
    auth: DomoAuth,
    alert_type: str,
    name: str,
    owner_id: int,
    resource_id: str,
    data_source_id: str,
    *,
    resource_type: str = "CARD",
    resource_name: str = "",
    operation: str = "GREATER_THAN",
    value: str = "",
    body: dict[str, Any] | None = None,
    return_raw: bool = False,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> DomoAlert | ResponseGetData

Create a new alert and return the hydrated :class:DomoAlert.

Builds the alert payload with alert_builder.build_alert_payload (or accepts a pre-built body for advanced alert types) and calls the create_alert route.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	DomoAuth instance.	required
alert_type	str	Friendly alert-type name (e.g. "summary_number"). See alert_builder.get_registered_alert_types().	required
name	str	Alert name.	required
owner_id	int	Domo user id of the alert owner.	required
resource_id	str	Card or dataset id the alert monitors.	required
data_source_id	str	Underlying dataset id (for filter groups).	required
resource_type	str	"CARD" or "DATASET".	'CARD'
resource_name	str	Display name of the monitored resource.	''
operation	str	Comparison operator (e.g. "GREATER_THAN").	'GREATER_THAN'
value	str	Threshold value as a string.	''
body	dict[str, Any] | None	Escape hatch — a pre-built alert payload (from build_alert_payload(...)). When given, the other alert arguments are ignored.	None
return_raw	bool	Return the raw create response instead of a DomoAlert.	False
context	RouteContext | None	Optional RouteContext.	None

Source code in src/crew_dcs/classes/DomoAlert.py

@classmethod
async def create(
    cls,
    auth: DomoAuth,
    alert_type: str,
    name: str,
    owner_id: int,
    resource_id: str,
    data_source_id: str,
    *,
    resource_type: str = "CARD",
    resource_name: str = "",
    operation: str = "GREATER_THAN",
    value: str = "",
    body: dict[str, Any] | None = None,
    return_raw: bool = False,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> DomoAlert | ResponseGetData:
    """Create a new alert and return the hydrated :class:`DomoAlert`.

    Builds the alert payload with ``alert_builder.build_alert_payload`` (or
    accepts a pre-built ``body`` for advanced alert types) and calls the
    ``create_alert`` route.

    Args:
        auth: DomoAuth instance.
        alert_type: Friendly alert-type name (e.g. ``"summary_number"``).
            See ``alert_builder.get_registered_alert_types()``.
        name: Alert name.
        owner_id: Domo user id of the alert owner.
        resource_id: Card or dataset id the alert monitors.
        data_source_id: Underlying dataset id (for filter groups).
        resource_type: ``"CARD"`` or ``"DATASET"``.
        resource_name: Display name of the monitored resource.
        operation: Comparison operator (e.g. ``"GREATER_THAN"``).
        value: Threshold value as a string.
        body: Escape hatch — a pre-built alert payload (from
            ``build_alert_payload(...)``). When given, the other alert
            arguments are ignored.
        return_raw: Return the raw create response instead of a DomoAlert.
        context: Optional RouteContext.
    """
    from .alert_builder import build_alert_payload

    # create + hydrate reuse one pooled session via auth-level connection
    # reuse (RouteContext.reuse_session, default on).
    if body is None:
        body = build_alert_payload(
            alert_type,
            name=name,
            owner_id=owner_id,
            resource_type=resource_type,
            resource_id=resource_id,
            resource_name=resource_name,
            data_source_id=data_source_id,
            operation=operation,
            value=value,
        )

    res = await alert_routes.create_alert(auth=auth, body=body, context=context)

    if return_raw:
        return res

    resp = res.response if isinstance(res.response, dict) else {}
    alert_id = resp.get("id") or resp.get("alertId")
    if not alert_id:
        raise ValueError(
            f"create_alert response did not contain an alert id: {res.response!r}"
        )

    return await cls.get_by_id(auth=auth, alert_id=str(alert_id), context=context)

delete async

delete(
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Delete this alert.

Source code in src/crew_dcs/classes/DomoAlert.py

async def delete(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> ResponseGetData:
    """Delete this alert."""
    return await alert_routes.delete_alert(
        auth=self.auth, alert_id=self.id, context=context, **context_kwargs
    )

from_dict classmethod

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

Converts an alerts API response into a DomoAlert instance.

Source code in src/crew_dcs/classes/DomoAlert.py

@classmethod
def from_dict(cls, auth: DomoAuth, obj: dict[str, Any]) -> DomoAlert:
    """Converts an alerts API response into a DomoAlert instance."""
    return cls(
        auth=auth,
        id=str(obj.get("id")),
        name=obj.get("name"),
        owner_id=str(obj["owner"]) if obj.get("owner") is not None else None,
        enabled=obj.get("enabled"),
        raw=obj,
    )

get_by_id async classmethod

get_by_id(
    auth: DomoAuth,
    alert_id: int | str,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> DomoAlert | ResponseGetData

Retrieve a single alert by ID.

Source code in src/crew_dcs/classes/DomoAlert.py

@classmethod
async def get_by_id(
    cls,
    auth: DomoAuth,
    alert_id: int | str,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> DomoAlert | ResponseGetData:
    """Retrieve a single alert by ID."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await alert_routes.get_alert(
        auth=auth,
        alert_id=str(alert_id),
        context=context,
    )

    if return_raw:
        return res

    return cls.from_dict(auth=auth, obj=res.response)

get_entity_by_id async classmethod

get_entity_by_id(
    entity_id: str, auth: DomoAuth, **kwargs: Any
) -> DomoAlert

Internal wrapper used by base refresh() method.

Source code in src/crew_dcs/classes/DomoAlert.py

@classmethod
async def get_entity_by_id(
    cls,
    entity_id: str,
    auth: DomoAuth,
    **kwargs: Any,
) -> DomoAlert:
    """Internal wrapper used by base refresh() method."""
    return await cls.get_by_id(alert_id=entity_id, auth=auth, **kwargs)

update_owner async

update_owner(
    to_user_id: int | str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> DomoAlert

Change the owner of this alert. Updates self.owner_id in place.

Source code in src/crew_dcs/classes/DomoAlert.py

async def update_owner(
    self,
    to_user_id: int | str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> DomoAlert:
    """Change the owner of this alert. Updates self.owner_id in place."""
    await alert_routes.update_alert_owner(
        auth=self.auth,
        alert_id=self.id,
        to_user_id=to_user_id,
        context=context,
        **context_kwargs,
    )
    self.owner_id = str(to_user_id)
    return self

DomoAlerts dataclass

DomoAlerts(auth: DomoAuth)

Bases: DomoManager

Manager for DomoAlert collections.

get async

get(
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> list[DomoAlert] | ResponseGetData

List all alerts visible to the authenticated user.

Source code in src/crew_dcs/classes/DomoAlert.py

async def get(
    self,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> list[DomoAlert] | ResponseGetData:
    """List all alerts visible to the authenticated user."""
    res = await alert_routes.list_alerts(
        auth=self.auth, context=context, **context_kwargs
    )

    if return_raw:
        return res

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

transfer_to_user async

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

Transfer all alerts owned by from_user_id to to_user_id.

Source code in src/crew_dcs/classes/DomoAlert.py

async def transfer_to_user(
    self,
    from_user_id: int | str,
    to_user_id: int | str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> list[ResponseGetData]:
    """Transfer all alerts owned by from_user_id to to_user_id."""
    return await alert_routes.transfer_alerts(
        auth=self.auth,
        from_user_id=from_user_id,
        to_user_id=to_user_id,
        context=context,
        **context_kwargs,
    )