alert_builder

alert_builder

Builder API for constructing Domo alert payloads from scratch.

This mirrors the kpi_builder pattern: a registry keyed by a friendly alert-type name, a @register_alert_type decorator, a polymorphic base builder, and concrete per-type subclasses.

Each builder accepts high-level parameters (name, resource, threshold, ...) and produces a dict ready for crew_dcs.routes.alerts.crud.create_alert.

Usage::

from crew_dcs.classes.alert_builder import build_alert_payload
from crew_dcs.routes.alerts.crud import create_alert

body = build_alert_payload(
    "summary_number",
    name="Revenue Alert",
    owner_id=1351541814,
    resource_type="CARD",
    resource_id="533357453",
    resource_name="Total Revenue (Actual)",
    data_source_id="c2132334-ee01-4728-9278-f9798a8d7273",
    operation="GREATER_THAN",
    value="1000000",
)
res = await create_alert(auth, body=body, context=ctx)

Alert types discovered from 50 existing alerts on domo-community:

============== =========== ===================================================
Friendly name  Domo type   Configurations
============== =========== ===================================================
summary_number SUMMARY_NUMBER  OPERATION, VALUE
aggregation    AGGREGATION     GROUP, AGGREGATION_TYPE, AGGREGATION, OPERATION, VALUE
any_row        ANY_ROW         OPERATION, ANY_ROW_PRIMARY_KEYS, ANY_ROW_METADATA_COLUMNS
column         COLUMN          COLUMN_NAME, COLUMN_ID, COLUMN_STATISTIC, OPERATION, VALUE
============== =========== ===================================================

AggregationAlertBuilder dataclass

AggregationAlertBuilder(
    name: str = "",
    owner_id: int = 0,
    resource_type: str = "CARD",
    resource_id: str = "",
    resource_name: str = "",
    data_source_id: str = "",
    operation: str = "GREATER_THAN",
    value: str = "",
    enabled: bool = True,
    notify_trigger: bool = True,
    notify_repeatable: bool = True,
    trigger_frequency: str = "Rarely",
    subscriptions: list[dict[str, Any]] = list(),
    group: str = "All Items",
    aggregation_type: str = "ALL",
    aggregation: str = "SUM",
)

Bases: AlertBuilder

Alert on an aggregation of a card's data crossing a threshold.

Useful for: "Alert when total expenses exceed budget" or "Alert when sum of all accounts exceeds $X".

Attributes:

Name	Type	Description
group	str	Group name (default "All Items")
aggregation_type	str	Aggregation scope (default "ALL")
aggregation	str	Aggregation function (default "SUM")
operation	str	GREATER_THAN, LESS_THAN, etc.
value	str	Threshold value as a string

AlertBuilder dataclass

AlertBuilder(
    name: str = "",
    owner_id: int = 0,
    resource_type: str = "CARD",
    resource_id: str = "",
    resource_name: str = "",
    data_source_id: str = "",
    operation: str = "GREATER_THAN",
    value: str = "",
    enabled: bool = True,
    notify_trigger: bool = True,
    notify_repeatable: bool = True,
    trigger_frequency: str = "Rarely",
    subscriptions: list[dict[str, Any]] = list(),
)

Base builder for a Domo alert payload.

Subclasses register themselves with :func:register_alert_type.

Attributes:

Name	Type	Description
name	str	Alert name (shown in Domo UI).
owner_id	int	Domo user ID of the alert owner.
resource_type	str	"CARD" or "DATASET".
resource_id	str	ID of the card or dataset the alert monitors.
resource_name	str	Display name of the resource.
data_source_id	str	The underlying data source ID for filter groups. For CARD alerts, this is the dataset backing the card. For DATASET alerts, this is the dataset ID itself.
operation	str	Comparison operator (e.g. "GREATER_THAN", "LESS_THAN", "GREATER_THAN_EQUAL", "CHANGES_BY", "ROWS_ADDED").
value	str	Threshold value as a string (e.g. "1000000").
enabled	bool	Whether the alert is active (default True).
notify_trigger	bool	Whether to send a notification on trigger (default True).
notify_repeatable	bool	Whether to repeat notifications (default True).
trigger_frequency	str	How often the alert is evaluated (e.g. "Rarely", "6 per year").
subscriptions	list[dict[str, Any]]	List of subscription dicts (who gets notified). If empty, the owner is auto-subscribed.

build

build() -> dict[str, Any]

Assemble and return the alert creation payload dict.

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

def build(self) -> dict[str, Any]:
    """Assemble and return the alert creation payload dict."""
    self._validate()

    return {
        "name": self.name,
        "type": self.alert_type,
        "owner": self.owner_id,
        "active": True,
        "enabled": self.enabled,
        "resourceType": self.resource_type,
        "resourceId": self.resource_id,
        "resourceName": self.resource_name,
        "triggered": False,
        "triggerFrequency": self.trigger_frequency,
        "configurations": self.build_configurations(),
        "filterGroups": self.build_filter_groups(),
        "contextual": False,
        "subscriptions": self.build_subscriptions(),
        "category": "DATA",
    }

build_configurations

build_configurations() -> list[dict[str, str]]

Build the type-specific configurations. Override in subclasses.

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

def build_configurations(self) -> list[dict[str, str]]:
    """Build the type-specific configurations. Override in subclasses."""
    return self._base_configurations()

build_filter_groups

build_filter_groups() -> list[dict[str, Any]]

Build the filter groups (default: one 'All Rows' group).

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

def build_filter_groups(self) -> list[dict[str, Any]]:
    """Build the filter groups (default: one 'All Rows' group)."""
    return [
        {
            "name": "All Rows",
            "filterGroupId": 0,
            "dataSourceId": self.data_source_id,
            "type": "open",
            "dataSourcePermissions": False,
            "order": 0,
        }
    ]

build_subscriptions

build_subscriptions() -> list[dict[str, Any]]

Build the subscription list. If empty, auto-subscribe the owner.

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

def build_subscriptions(self) -> list[dict[str, Any]]:
    """Build the subscription list. If empty, auto-subscribe the owner."""
    if self.subscriptions:
        return self.subscriptions
    if self.owner_id:
        return [
            {
                "subscriberId": str(self.owner_id),
                "type": "USER",
                "subscribedBy": self.owner_id,
            }
        ]
    return []

AnyRowAlertBuilder dataclass

AnyRowAlertBuilder(
    name: str = "",
    owner_id: int = 0,
    resource_type: str = "DATASET",
    resource_id: str = "",
    resource_name: str = "",
    data_source_id: str = "",
    operation: str = "ROWS_ADDED",
    value: str = "",
    enabled: bool = True,
    notify_trigger: bool = True,
    notify_repeatable: bool = True,
    trigger_frequency: str = "Rarely",
    subscriptions: list[dict[str, Any]] = list(),
    primary_keys: str = "",
    metadata_columns: str = "",
)

Bases: AlertBuilder

Alert when rows are added or changed in a dataset.

Useful for: "Alert when new transactions are loaded" or "Alert when any row is added to the GL dataset".

Attributes:

Name	Type	Description
operation	str	ROWS_ADDED (default)
primary_keys	str	Comma-separated column names that identify a row
metadata_columns	str	Comma-separated column names to include in the alert

ColumnAlertBuilder dataclass

ColumnAlertBuilder(
    name: str = "",
    owner_id: int = 0,
    resource_type: str = "DATASET",
    resource_id: str = "",
    resource_name: str = "",
    data_source_id: str = "",
    operation: str = "GREATER_THAN",
    value: str = "",
    enabled: bool = True,
    notify_trigger: bool = True,
    notify_repeatable: bool = True,
    trigger_frequency: str = "Rarely",
    subscriptions: list[dict[str, Any]] = list(),
    column_name: str = "",
    column_id: str = "",
    column_statistic: str = "UNIQUE_COUNT",
)

Bases: AlertBuilder

Alert on a column statistic crossing a threshold.

Useful for: "Alert when unique count of accounts changes" or "Alert when null count in a column exceeds N".

Attributes:

Name	Type	Description
column_name	str	Column name to monitor
column_id	str	Column ID (usually same as column_name)
column_statistic	str	Statistic to watch (UNIQUE_COUNT, NULL_COUNT, etc.)
operation	str	GREATER_THAN, LESS_THAN, CHANGES_BY, etc.
value	str	Threshold value as a string

SummaryNumberAlertBuilder dataclass

SummaryNumberAlertBuilder(
    name: str = "",
    owner_id: int = 0,
    resource_type: str = "CARD",
    resource_id: str = "",
    resource_name: str = "",
    data_source_id: str = "",
    operation: str = "GREATER_THAN",
    value: str = "",
    enabled: bool = True,
    notify_trigger: bool = True,
    notify_repeatable: bool = True,
    trigger_frequency: str = "Rarely",
    subscriptions: list[dict[str, Any]] = list(),
)

Bases: AlertBuilder

Alert on a card's summary number crossing a threshold.

The most common alert type for financial KPIs: "Alert when Net Income drops below $X" or "Alert when Revenue exceeds $Y".

Attributes:

Name	Type	Description
operation	str	GREATER_THAN, LESS_THAN, GREATER_THAN_EQUAL, LESS_THAN_EQUAL, CHANGES_BY
value	str	Threshold value as a string

alert_type_map

alert_type_map() -> dict[str, str]

Return the friendly-name -> Domo-alert-type mapping.

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

def alert_type_map() -> dict[str, str]:
    """Return the friendly-name -> Domo-alert-type mapping."""
    return dict(_ALERT_TYPE_MAP)

build_alert_payload

build_alert_payload(
    alert_type: str, **kwargs: Any
) -> dict[str, Any]

Build an alert creation payload for a friendly alert-type name.

Parameters:

Name	Type	Description	Default
alert_type	str	Friendly alert-type name (see :func:get_registered_alert_types).	required
**kwargs	Any	Passed through to the :class:AlertBuilder subclass.	{}

Returns:

Type	Description
dict[str, Any]	Dict ready for create_alert(auth, body=...).

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

def build_alert_payload(alert_type: str, **kwargs: Any) -> dict[str, Any]:
    """Build an alert creation payload for a friendly alert-type name.

    Args:
        alert_type: Friendly alert-type name (see
            :func:`get_registered_alert_types`).
        **kwargs: Passed through to the :class:`AlertBuilder` subclass.

    Returns:
        Dict ready for ``create_alert(auth, body=...)``.
    """
    builder_cls = get_alert_builder_class(alert_type)
    builder = builder_cls(**kwargs)
    return builder.build()

get_alert_builder_class

get_alert_builder_class(
    friendly_name: str,
) -> type[AlertBuilder]

Return the builder class registered for a friendly alert-type name.

Raises:

Type	Description
ValueError	If the friendly name is not registered.

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

def get_alert_builder_class(friendly_name: str) -> type[AlertBuilder]:
    """Return the builder class registered for a friendly alert-type name.

    Raises:
        ValueError: If the friendly name is not registered.
    """
    cls = _ALERT_BUILDER_REGISTRY.get(friendly_name)
    if cls is None:
        known = ", ".join(sorted(_ALERT_BUILDER_REGISTRY))
        raise ValueError(f"Unknown alert type '{friendly_name}'. Registered: {known}")
    return cls

get_registered_alert_types

get_registered_alert_types() -> list[str]

Return all registered friendly alert-type names.

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

def get_registered_alert_types() -> list[str]:
    """Return all registered friendly alert-type names."""
    return sorted(_ALERT_BUILDER_REGISTRY)

register_alert_type

register_alert_type(friendly_name: str, alert_type: str)

Decorator registering an :class:AlertBuilder subclass.

Parameters:

Name	Type	Description	Default
friendly_name	str	Human-friendly key (e.g. "summary_number").	required
alert_type	str	Domo alert type string (e.g. "SUMMARY_NUMBER").	required

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

def register_alert_type(friendly_name: str, alert_type: str):
    """Decorator registering an :class:`AlertBuilder` subclass.

    Args:
        friendly_name: Human-friendly key (e.g. ``"summary_number"``).
        alert_type: Domo alert ``type`` string (e.g. ``"SUMMARY_NUMBER"``).
    """

    def decorator(cls: type[AlertBuilder]) -> type[AlertBuilder]:
        cls.friendly_name = friendly_name
        cls.alert_type = alert_type
        _ALERT_BUILDER_REGISTRY[friendly_name] = cls
        _ALERT_TYPE_MAP[friendly_name] = alert_type
        return cls

    return decorator