Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions packages/gooddata-sdk/src/gooddata_sdk/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -210,6 +210,7 @@
CatalogDeclarativeExportDefinitionRequestPayload,
)
from gooddata_sdk.catalog.workspace.declarative_model.workspace.automation import (
CatalogAutomationAlert,
CatalogAutomationSchedule,
CatalogDeclarativeAutomation,
)
Expand Down
Original file line number Diff line number Diff line change
@@ -1,8 +1,11 @@
# (C) 2024 GoodData Corporation
from __future__ import annotations

import builtins
from typing import Any

from attrs import define, field
from gooddata_api_client.model.automation_alert import AutomationAlert
from gooddata_api_client.model.automation_schedule import AutomationSchedule
from gooddata_api_client.model.automation_tabular_export import AutomationTabularExport
from gooddata_api_client.model.automation_visual_export import AutomationVisualExport
Expand All @@ -21,6 +24,30 @@
from gooddata_sdk.catalog.workspace.declarative_model.workspace.analytics_model.base import CatalogAnalyticsBaseMeta


@define(kw_only=True)
class CatalogAutomationAlert(Base):
"""Wraps the alert configuration for an automation.

The ``trigger`` field controls when the alert fires:
- ``ALWAYS`` – fires every time the condition is met.
- ``ONCE`` – fires only the first time the condition is met.
- ``ONCE_PER_INTERVAL`` – fires when the condition is met, then is suppressed for the
duration specified by ``interval``.

The ``interval`` field is only used when ``trigger`` is ``ONCE_PER_INTERVAL`` and
must be one of: ``DAY``, ``WEEK``, ``MONTH``, ``QUARTER``, ``YEAR``.
"""

condition: dict[str, Any]
execution: dict[str, Any]
trigger: str | None = None
interval: str | None = None

@staticmethod
def client_class() -> builtins.type[AutomationAlert]:
return AutomationAlert


@define(kw_only=True)
class CatalogAutomationSchedule(Base):
cron: str
Expand Down Expand Up @@ -65,6 +92,7 @@ class CatalogDeclarativeAutomation(CatalogAnalyticsBaseMeta):
schedule: CatalogAutomationSchedule | None = None
tabular_exports: list[CatalogAutomationTabularExport] | None = None
visual_exports: list[CatalogAutomationVisualExport] | None = None
alert: CatalogAutomationAlert | None = None

@staticmethod
def client_class() -> builtins.type[DeclarativeAutomation]:
Expand Down
39 changes: 39 additions & 0 deletions packages/gooddata-sdk/tests/catalog/test_catalog_workspace.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import yaml
from gooddata_sdk import (
BasicCredentials,
CatalogAutomationAlert,
CatalogAutomationSchedule,
CatalogDataSourcePostgres,
CatalogDeclarativeAutomation,
Expand Down Expand Up @@ -1032,3 +1033,41 @@ def test_layout_filter_views(test_config):
assert filter_views_expected == filter_views_o
finally:
safe_delete(sdk.catalog_workspace.put_declarative_filter_views, workspace_id, [])


def test_catalog_automation_alert_once_per_interval():
"""Unit test: CatalogAutomationAlert round-trips ONCE_PER_INTERVAL + interval correctly."""
condition = {"type": "comparison", "operator": "GREATER_THAN", "right": {"value": 100}}
execution = {"measures": [{"localIdentifier": "m1", "definition": {"measure": {"item": {"identifier": {"id": "revenue", "type": "measure"}}}}}]}

alert = CatalogAutomationAlert(
condition=condition,
execution=execution,
trigger="ONCE_PER_INTERVAL",
interval="MONTH",
)

assert alert.trigger == "ONCE_PER_INTERVAL"
assert alert.interval == "MONTH"

# to_api should produce an AutomationAlert with interval and trigger set
api_model = alert.to_api()
api_dict = api_model.to_dict()
assert api_dict.get("trigger") == "ONCE_PER_INTERVAL"
assert api_dict.get("interval") == "MONTH"


def test_catalog_automation_alert_defaults():
"""Unit test: CatalogAutomationAlert trigger and interval are optional (None by default)."""
condition = {"type": "comparison", "operator": "GREATER_THAN", "right": {"value": 0}}
execution = {"measures": []}

alert = CatalogAutomationAlert(condition=condition, execution=execution)

assert alert.trigger is None
assert alert.interval is None

api_dict = alert.to_api().to_dict()
# trigger/interval should not appear when not set
assert "trigger" not in api_dict or api_dict.get("trigger") is None
assert "interval" not in api_dict or api_dict.get("interval") is None
Loading