Skip to content

Schedules

Schedule dataclass

Schedule(
    kind: str,
    name: str | None = None,
    kwargs: dict = dict(),
    enabled: bool = True,
    cron: str | None = None,
    interval_minutes: int | None = None,
    repeats: int | None = None,
    run_at: datetime | None = None,
)

Frozen value object describing one declarative schedule.

name identifies the schedule for the startup sync (identity is (command_name, source=code, name)); a single declaration defaults its name to the command name. enabled is only the initial value — after the first sync the database owns it (admins may pause without a deploy).

model_kwargs

model_kwargs() -> dict

Field values for the ScheduledCommand row (spec only).

Source code in src/django_admin_runner/schedules.py
def model_kwargs(self) -> dict:
    """Field values for the ``ScheduledCommand`` row (spec only)."""
    return {
        "kind": self.kind,
        "cron": self.cron or "",
        "interval_minutes": self.interval_minutes,
        "repeats": self.repeats,
        "run_at": self.run_at,
        "kwargs": dict(self.kwargs),
    }

CronSchedule

CronSchedule(cron, *, name=None, kwargs=None, enabled=True)

Bases: Schedule

Run on a cron expression, e.g. "15 4 * * *".

Source code in src/django_admin_runner/schedules.py
def __init__(self, cron, *, name=None, kwargs=None, enabled=True):
    validate_cron_expression(cron)
    super().__init__(
        "cron",
        name=name,
        kwargs=dict(kwargs or {}),
        enabled=enabled,
        cron=cron,
    )

IntervalSchedule

IntervalSchedule(
    minutes,
    *,
    repeats=None,
    name=None,
    kwargs=None,
    enabled=True,
)

Bases: Schedule

Run every minutes minutes, optionally limited to repeats runs.

Source code in src/django_admin_runner/schedules.py
def __init__(self, minutes, *, repeats=None, name=None, kwargs=None, enabled=True):
    minutes = int(minutes)
    if minutes <= 0:
        raise ValueError(f"Interval minutes must be positive, got {minutes}.")
    if repeats is not None:
        repeats = int(repeats)
        if repeats <= 0:
            raise ValueError(f"Repeats must be positive, got {repeats}.")
    super().__init__(
        "interval",
        name=name,
        kwargs=dict(kwargs or {}),
        enabled=enabled,
        interval_minutes=minutes,
        repeats=repeats,
    )

ClockedSchedule

ClockedSchedule(
    at, *, name=None, kwargs=None, enabled=True
)

Bases: Schedule

Run once at a specific (aware) datetime.

Source code in src/django_admin_runner/schedules.py
def __init__(self, at, *, name=None, kwargs=None, enabled=True):
    if not isinstance(at, datetime):
        raise TypeError("ClockedSchedule 'at' must be a datetime.")
    if at.tzinfo is None:
        at = at.replace(tzinfo=UTC)
    if at <= datetime.now(UTC):
        raise ValueError("ClockedSchedule 'at' must be in the future.")
    super().__init__(
        "clocked",
        name=name,
        kwargs=dict(kwargs or {}),
        enabled=enabled,
        run_at=at,
    )

validate_cron_expression

validate_cron_expression(expression: str) -> None

Validate a cron expression; raises ValueError when unparseable.

Prefers croniter (shipped with django-q2); falls back to a basic five-field check when it is not installed.

Source code in src/django_admin_runner/schedules.py
def validate_cron_expression(expression: str) -> None:
    """Validate a cron expression; raises ``ValueError`` when unparseable.

    Prefers ``croniter`` (shipped with django-q2); falls back to a basic
    five-field check when it is not installed.
    """
    try:
        from croniter import croniter

        croniter(expression)
        return
    except ImportError:
        pass
    parts = expression.split()
    if len(parts) != 5 or not all(re.fullmatch(r"[\d*/,\-]+", p) for p in parts):
        raise ValueError(f"Invalid cron expression: {expression!r}")