schedule

schedule

A class-based approach for interpreting Domo schedule configurations

DomoCronSchedule dataclass

DomoCronSchedule(
    schedule_start_date: datetime | None = None,
    frequency: ScheduleFrequencyEnum = MANUAL,
    schedule_type: ScheduleType = MANUAL,
    interval: int = 1,
    minute: int | None = None,
    hour: int | None = None,
    day_of_week: list[int] | None = None,
    day_of_month: list[int] | None = None,
    month: list[int] | None = None,
    timezone: str | None = None,
    is_active: bool = True,
    next_run_time: datetime | None = None,
    raw: dict[str, Any] = dict(),
    parent: Any = None,
    schedule_expression: str | None = None,
    advanced_schedule_json: dict[str, Any] | None = None,
)

Bases: DomoSchedule_Base

Schedule based on cron-like expressions, advanced JSON, or simple expressions

from_dict classmethod

from_dict(
    obj: dict[str, Any], parent: Any | None = None, **kwargs
) -> DomoCronSchedule

Create DomoCronSchedule from dictionary/API response

Source code in src/crew_dcs/classes/subentity/schedule.py

@classmethod
def from_dict(
    cls,
    obj: dict[str, Any],
    parent: Any | None = None,
    **kwargs,
) -> "DomoCronSchedule":
    """Create DomoCronSchedule from dictionary/API response"""
    field_mappings = cls._extract_field_mappings(obj)
    start_date = cls._parse_datetime_input(field_mappings["start_date_raw"])
    advanced_json = cls._parse_json_input(field_mappings["advanced_json"])

    # Extract only scheduler-relevant fields for raw storage
    scheduler_raw = cls._extract_scheduler_fields(obj)

    return cls(
        parent=parent,
        schedule_start_date=start_date,
        schedule_expression=field_mappings["schedule_expr"],
        advanced_schedule_json=advanced_json,
        timezone=field_mappings["timezone"],
        is_active=field_mappings["is_active"],
        raw=scheduler_raw,
        **kwargs,
    )

DomoManualSchedule dataclass

DomoManualSchedule(
    schedule_start_date: datetime | None = None,
    frequency: ScheduleFrequencyEnum = MANUAL,
    schedule_type: ScheduleType = MANUAL,
    interval: int = 1,
    minute: int | None = None,
    hour: int | None = None,
    day_of_week: list[int] | None = None,
    day_of_month: list[int] | None = None,
    month: list[int] | None = None,
    timezone: str | None = None,
    is_active: bool = True,
    next_run_time: datetime | None = None,
    raw: dict[str, Any] = dict(),
    parent: Any = None,
    schedule_expression: str | None = None,
)

Bases: DomoSchedule_Base

Schedule for manual execution only

from_dict classmethod

from_dict(
    obj: dict[str, Any], parent: Any = None, **kwargs
) -> DomoManualSchedule

Create DomoManualSchedule from dictionary/API response

Source code in src/crew_dcs/classes/subentity/schedule.py

@classmethod
def from_dict(
    cls, obj: dict[str, Any], parent: Any = None, **kwargs
) -> "DomoManualSchedule":
    """Create DomoManualSchedule from dictionary/API response"""
    field_mappings = cls._extract_field_mappings(obj)
    start_date = cls._parse_datetime_input(field_mappings["start_date_raw"])

    # Extract only scheduler-relevant fields for raw storage
    scheduler_raw = cls._extract_scheduler_fields(obj)

    return cls(
        parent=parent,
        schedule_start_date=start_date,
        schedule_expression=field_mappings["schedule_expr"],
        timezone=field_mappings["timezone"],
        is_active=field_mappings["is_active"],
        raw=scheduler_raw,
        **kwargs,
    )

DomoSchedule

Factory for schedule parsing.

Some entities (e.g. Streams, Triggers) treat schedule configuration as an attached subentity but need a consistent entry point.

determine_schedule_type staticmethod

determine_schedule_type(
    obj: dict[str, Any] | None,
) -> type[DomoSchedule_Base]

Determine the appropriate schedule subclass for the given raw payload.

Source code in src/crew_dcs/classes/subentity/schedule.py

@staticmethod
def determine_schedule_type(obj: dict[str, Any] | None) -> type[DomoSchedule_Base]:
    """Determine the appropriate schedule subclass for the given raw payload."""
    if not obj:
        return DomoManualSchedule

    field_mappings = DomoSchedule_Base._extract_field_mappings(obj)

    advanced_json = DomoSchedule_Base._parse_json_input(
        field_mappings["advanced_json"]
    )
    if advanced_json:
        # Advanced schedule present → cron-like schedule.
        # Manual schedules generally do not include advancedScheduleJson.
        return DomoCronSchedule

    expr = field_mappings["schedule_expr"]
    if expr:
        # Any non-empty expression should be parsed as a cron-like schedule.
        return DomoCronSchedule

    return DomoManualSchedule

from_dict classmethod

from_dict(
    obj: dict[str, Any] | None,
    parent: Any | None = None,
    **kwargs
) -> DomoSchedule_Base

Create a schedule instance from a raw dict.

Source code in src/crew_dcs/classes/subentity/schedule.py

@classmethod
def from_dict(
    cls,
    obj: dict[str, Any] | None,
    parent: Any | None = None,
    **kwargs,
) -> DomoSchedule_Base:
    """Create a schedule instance from a raw dict."""
    schedule_cls = cls.determine_schedule_type(obj)
    if obj is None:
        obj = {}
    return schedule_cls.from_dict(obj=obj, parent=parent, **kwargs)

from_parent classmethod

from_parent(
    parent: Any, obj: dict[str, Any] | None = None, **kwargs
) -> DomoSchedule_Base

Create a schedule instance from a parent entity and (optional) raw payload.

Source code in src/crew_dcs/classes/subentity/schedule.py

@classmethod
def from_parent(
    cls,
    parent: Any,
    obj: dict[str, Any] | None = None,
    **kwargs,
) -> DomoSchedule_Base:
    """Create a schedule instance from a parent entity and (optional) raw payload."""
    if obj is None:
        obj = getattr(parent, "raw", None) or {}
    return cls.from_dict(obj=obj, parent=parent, **kwargs)

DomoSchedule_Base dataclass

DomoSchedule_Base(
    schedule_start_date: datetime | None = None,
    frequency: ScheduleFrequencyEnum = MANUAL,
    schedule_type: ScheduleType = MANUAL,
    interval: int = 1,
    minute: int | None = None,
    hour: int | None = None,
    day_of_week: list[int] | None = None,
    day_of_month: list[int] | None = None,
    month: list[int] | None = None,
    timezone: str | None = None,
    is_active: bool = True,
    next_run_time: datetime | None = None,
    raw: dict[str, Any] = dict(),
    parent: Any = None,
)

Bases: DomoBase, ABC

Base class for interpreting and managing Domo schedule configurations

is_manual property

is_manual: bool

True when this schedule resolves to manual execution.

Checks advanced_schedule_json["type"] when present, before falling back to schedule_expression — a schedule built entirely from advancedScheduleJson (no plain cron string) has no schedule_expression, and that field defaulting to "" used to match the manual-marker set below unconditionally, so a dataset scheduled via advancedScheduleJson always read as MANUAL regardless of what type it actually carried (issue #1505).

is_scheduled property

is_scheduled: bool

True when this schedule is not manual execution.

compute_next_run_time

compute_next_run_time(
    after_utc: datetime | None = None,
) -> datetime | None

Compute the next fire time for this schedule's cron expression.

Domo does not populate nextRunTime on live triggers (issue

1505), so this computes it from schedule_expression instead of

reading a field Domo never sends. Only a plain Quartz-shaped schedule_expression (6 or 7 whitespace-separated fields) is parsed here; a schedule with no expression, a non-cron simple expression (e.g. "DAILY"), or an advanced-JSON-only schedule returns None — advancedScheduleJson is a different, non-cron -string shape this parser does not attempt to guess at.

Returns a timezone-aware UTC datetime, computed DST-correctly via zoneinfo against self.timezone (defaulting to UTC).

Raises:

Type	Description
QuartzCronParseError	schedule_expression looks like Quartz cron (right field count) but fails to parse — a malformed schedule must fail loudly, not silently read as unscheduled.

Source code in src/crew_dcs/classes/subentity/schedule.py

def compute_next_run_time(
    self, after_utc: dt.datetime | None = None
) -> dt.datetime | None:
    """Compute the next fire time for this schedule's cron expression.

    Domo does not populate `nextRunTime` on live triggers (issue
    #1505), so this computes it from `schedule_expression` instead of
    reading a field Domo never sends. Only a plain Quartz-shaped
    `schedule_expression` (6 or 7 whitespace-separated fields) is
    parsed here; a schedule with no expression, a non-cron simple
    expression (e.g. `"DAILY"`), or an advanced-JSON-only schedule
    returns `None` — `advancedScheduleJson` is a different, non-cron
    -string shape this parser does not attempt to guess at.

    Returns a timezone-aware UTC `datetime`, computed DST-correctly via
    `zoneinfo` against `self.timezone` (defaulting to UTC).

    Raises:
        QuartzCronParseError: `schedule_expression` looks like Quartz
            cron (right field count) but fails to parse — a malformed
            schedule must fail loudly, not silently read as unscheduled.
    """
    expression = getattr(self, "schedule_expression", None)
    if not expression:
        return None
    if len(expression.split()) not in (6, 7):
        return None

    return quartz_cron.compute_next_run_time(
        expression, self.timezone, after_utc=after_utc
    )

export_as_dict

export_as_dict(
    override_fn: Callable | None = None,
    return_snake_case: bool = False,
) -> dict[str, Any]

Export the schedule as a unified dictionary format. Args: override_fn: Optional function to override export logic. return_snake_case: If True, return keys in snake_case format. Returns: dict: Unified schedule dictionary.

Source code in src/crew_dcs/classes/subentity/schedule.py

def export_as_dict(
    self, override_fn: Callable | None = None, return_snake_case: bool = False
) -> dict[str, Any]:
    """
    Export the schedule as a unified dictionary format.
    Args:
        override_fn: Optional function to override export logic.
        return_snake_case: If True, return keys in snake_case format.
    Returns:
        dict: Unified schedule dictionary.
    """
    if override_fn:
        return override_fn(self)

    # Unified schedule dict
    result = {
        "humanReadable": self.get_human_readable_schedule(),
        "frequency": self.frequency.value if self.frequency else None,
        "scheduleType": (self.schedule_type.value if self.schedule_type else None),
        "interval": self.interval,
        "minute": self.minute,
        "hour": self.hour,
        "dayOfWeek": self.day_of_week,
        "dayOfMonth": self.day_of_month,
        "month": self.month,
        "scheduleStartDate": (
            self.schedule_start_date.isoformat()
            if self.schedule_start_date
            else None
        ),
        "timezone": self.timezone,
        "isActive": self.is_active,
        "nextRunTime": (
            self.next_run_time.isoformat() if self.next_run_time else None
        ),
    }

    # Optionally include raw expression or advanced config if present
    if hasattr(self, "schedule_expression") and self.schedule_expression:
        result["expression"] = self.schedule_expression
    if hasattr(self, "advanced_schedule_json") and self.advanced_schedule_json:
        result["advancedScheduleJson"] = self.advanced_schedule_json

    # Convert to snake_case if requested
    if return_snake_case:
        from ...utils.convert import convert_str_to_snake_case

        result = {
            convert_str_to_snake_case(k, is_pascal=True): v
            for k, v in result.items()
        }

    return result

get_human_readable_schedule

get_human_readable_schedule() -> str

Get a human-readable description of the schedule

Source code in src/crew_dcs/classes/subentity/schedule.py

def get_human_readable_schedule(self) -> str:  # noqa: C901
    """Get a human-readable description of the schedule"""
    # Check if advanced_schedule_json indicates a different actual type
    if hasattr(self, "advanced_schedule_json") and self.advanced_schedule_json:
        actual_type = self.advanced_schedule_json.get("type", "").upper()
        if (
            actual_type == "MANUAL"
            and self.frequency != ScheduleFrequencyEnum.MANUAL
        ):
            return f"Manual execution (configured as {self.frequency.value})"

    if self.frequency == ScheduleFrequencyEnum.MANUAL:
        return "Manual execution"

    if self.frequency == ScheduleFrequencyEnum.ONCE:
        if self.schedule_start_date:
            return (
                f"Run once on {self.schedule_start_date.strftime('%Y-%m-%d %H:%M')}"
            )
        return "Run once"

    base_desc = ""

    if self.frequency == ScheduleFrequencyEnum.MINUTELY:
        base_desc = f"Every {self.interval} minute(s)"
    elif self.frequency == ScheduleFrequencyEnum.HOURLY:
        base_desc = f"Every {self.interval} hour(s)"
        if self.minute is not None:
            base_desc += f" at {self.minute:02d} minutes past the hour"
    elif self.frequency == ScheduleFrequencyEnum.DAILY:
        base_desc = f"Every {self.interval} day(s)"
        if self.hour is not None and self.minute is not None:
            base_desc += f" at {self.hour:02d}:{self.minute:02d}"
    elif self.frequency == ScheduleFrequencyEnum.WEEKLY:
        base_desc = f"Every {self.interval} week(s)"
        if self.day_of_week:
            days = [
                "Sunday",
                "Monday",
                "Tuesday",
                "Wednesday",
                "Thursday",
                "Friday",
                "Saturday",
            ]
            day_names = [days[day] for day in self.day_of_week if 0 <= day <= 6]
            base_desc += f" on {', '.join(day_names)}"
        if self.hour is not None and self.minute is not None:
            base_desc += f" at {self.hour:02d}:{self.minute:02d}"
    elif self.frequency == ScheduleFrequencyEnum.MONTHLY:
        base_desc = f"Every {self.interval} month(s)"
        if self.day_of_month:
            base_desc += f" on day(s) {', '.join(map(str, self.day_of_month))}"
        if self.hour is not None and self.minute is not None:
            base_desc += f" at {self.hour:02d}:{self.minute:02d}"
    elif self.frequency == ScheduleFrequencyEnum.YEARLY:
        base_desc = f"Every {self.interval} year(s)"
        if self.month:
            month_names = [
                "Jan",
                "Feb",
                "Mar",
                "Apr",
                "May",
                "Jun",
                "Jul",
                "Aug",
                "Sep",
                "Oct",
                "Nov",
                "Dec",
            ]
            month_names_list = [
                month_names[m - 1] for m in self.month if 1 <= m <= 12
            ]
            base_desc += f" in {', '.join(month_names_list)}"
        if self.day_of_month:
            base_desc += f" on day(s) {', '.join(map(str, self.day_of_month))}"
    elif self.frequency == ScheduleFrequencyEnum.CUSTOM_CRON:
        base_desc = "Custom schedule"

    if self.timezone:
        base_desc += f" ({self.timezone})"

    if not self.is_active:
        base_desc += " [INACTIVE]"

    return base_desc

is_due_now

is_due_now(current_time: datetime | None = None) -> bool

Check if the schedule is due to run now (simplified logic)

Source code in src/crew_dcs/classes/subentity/schedule.py

def is_due_now(self, current_time: dt.datetime | None = None) -> bool:
    """Check if the schedule is due to run now (simplified logic)"""
    if not self.is_active or self.frequency == ScheduleFrequencyEnum.MANUAL:
        return False

    if current_time is None:
        current_time = dt.datetime.now()

    if self.schedule_start_date and current_time < self.schedule_start_date:
        return False

    # This is a simplified check - real implementation would need more complex logic
    # based on the specific schedule configuration
    return False  # Placeholder

refresh async

refresh(
    session: Any | None = None,
    debug_api: bool = False,
    is_suppress_no_config: bool = False,
    **kwargs
) -> DomoSchedule_Base

Refresh schedule data by refreshing the parent entity.

This method calls the parent entity's refresh() method, which updates the parent's raw data. The schedule then re-initializes itself from the updated parent data.

Note: If the schedule type changes (e.g., Manual to Cron), this method updates the instance in place. The parent entity should recreate the Schedule subentity if type safety is critical.

Parameters:

Name	Type	Description	Default
session	Any | None	Optional httpx.AsyncClient session for API requests	None
debug_api	bool	Enable debug output for API calls	False
is_suppress_no_config	bool	Suppress errors for missing configs	False

Returns:

Type	Description
DomoSchedule_Base	Self (with updated schedule data)

Raises:

Type	Description
AttributeError	If parent is None or doesn't support refresh

Source code in src/crew_dcs/classes/subentity/schedule.py

async def refresh(
    self,
    session: Any | None = None,
    debug_api: bool = False,
    is_suppress_no_config: bool = False,
    **kwargs,
) -> "DomoSchedule_Base":
    """Refresh schedule data by refreshing the parent entity.

    This method calls the parent entity's refresh() method, which updates
    the parent's raw data. The schedule then re-initializes itself from
    the updated parent data.

    Note: If the schedule type changes (e.g., Manual to Cron), this method
    updates the instance in place. The parent entity should recreate the
    Schedule subentity if type safety is critical.

    Args:
        session: Optional httpx.AsyncClient session for API requests
        debug_api: Enable debug output for API calls
        is_suppress_no_config: Suppress errors for missing configs

    Returns:
        Self (with updated schedule data)

    Raises:
        AttributeError: If parent is None or doesn't support refresh
    """
    if not self.parent:
        raise AttributeError("Cannot refresh schedule: parent entity is not set")

    # Refresh the parent entity
    if hasattr(self.parent, "refresh") and callable(self.parent.refresh):
        await self.parent.refresh(
            session=session,
            debug_api=debug_api,
            is_suppress_no_config=is_suppress_no_config,
            **kwargs,
        )
    else:
        raise AttributeError(
            f"Parent entity {type(self.parent).__name__} does not have a refresh() method"
        )

    # Extract updated schedule data from parent's raw response
    parent_raw = getattr(self.parent, "raw", {})

    # Get scheduler fields from parent
    scheduler_raw = self._extract_scheduler_fields(parent_raw)
    self.raw = scheduler_raw

    # Re-extract and parse the updated schedule components
    field_mappings = self._extract_field_mappings(parent_raw)

    # Update schedule start date
    self.schedule_start_date = self._parse_datetime_input(
        field_mappings["start_date_raw"]
    )

    # Update basic fields
    self.timezone = field_mappings["timezone"]
    self.is_active = field_mappings["is_active"]

    # Update schedule expression if present
    if hasattr(self, "schedule_expression"):
        self.schedule_expression = field_mappings["schedule_expr"]  # type: ignore

    # Update advanced JSON if present - this may cause type to change
    advanced_json_raw = field_mappings["advanced_json"]
    if hasattr(self, "advanced_schedule_json"):
        self.advanced_schedule_json = self._parse_json_input(  # type: ignore
            advanced_json_raw
        )

    # Re-interpret the schedule to update frequency, interval, etc.
    # Note: For DomoManualSchedule, this will still set to MANUAL
    # For DomoCronSchedule, this will re-parse the cron/advanced config
    self._interpret_schedule()

    return self

to_dict

to_dict(
    override_fn: Callable | None = None,
    return_snake_case: bool = False,
) -> dict[str, Any]

Convert schedule to dictionary format.

Source code in src/crew_dcs/classes/subentity/schedule.py

def to_dict(
    self, override_fn: Callable | None = None, return_snake_case: bool = False
) -> dict[str, Any]:
    """Convert schedule to dictionary format."""
    if override_fn:
        return override_fn(self)

    result = super().to_dict(return_snake_case=return_snake_case)

    schedule_type_key = "schedule_type" if return_snake_case else "scheduleType"
    is_active_key = "is_active" if return_snake_case else "isActive"

    result.update(
        {
            "frequency": self.frequency.value,
            schedule_type_key: self.schedule_type.value,
            "interval": self.interval,
            is_active_key: self.is_active,
        }
    )

    schedule_start_date_key = (
        "schedule_start_date" if return_snake_case else "scheduleStartDate"
    )
    if self.schedule_start_date:
        result[schedule_start_date_key] = self.schedule_start_date.isoformat()

    day_of_week_key = "day_of_week" if return_snake_case else "dayOfWeek"
    if self.day_of_week is not None:
        result[day_of_week_key] = self.day_of_week

    day_of_month_key = "day_of_month" if return_snake_case else "dayOfMonth"
    if self.day_of_month is not None:
        result[day_of_month_key] = self.day_of_month

    next_run_time_key = "next_run_time" if return_snake_case else "nextRunTime"
    if self.next_run_time:
        result[next_run_time_key] = self.next_run_time.isoformat()

    return result

ScheduleFrequencyEnum

Bases: DomoEnumMixin, Enum

Common schedule frequency types

ScheduleType

Bases: DomoEnumMixin, Enum

Schedule configuration types