Schedule

Build a schedule object for batch/workflow jobs using friendly fields.

Examples:

schedule = Schedule(at_hours=[0])
schedule = Schedule(at_minutes=[0, 30], weekdays=["Mon", "Wed", "Fri"])

You can either provide a raw cron_expression or use the human-readable helper parameters (which are compiled into cron syntax automatically).

Parameters:

• cron_expression

  (str, default: '' ) –

  A raw cron string (e.g. "0 */2 * * *"). If provided alongside helper parameters, the helpers take precedence.

• at_minutes

  (list[int] | None, default: None ) –

  Specific minutes past the hour to run (e.g. [0, 30] for ":00" and ":30").

• every_minute

  (int | None, default: None ) –

  Run every N minutes (e.g. 5 for every 5 minutes).

• at_hours

  (list[int] | None, default: None ) –

  Specific hours of the day to run (0 -- 23).

• every_hour

  (int | None, default: None ) –

  Run every N hours.

• weekdays

  (list[str] | None, default: None ) –

  Days of the week to run (e.g. ["mon", "wed", "fri"]).

• day_of_month

  (int | None, default: None ) –

  Day of the month to run (1 -- 31).

• in_month

  (list[str] | None, default: None ) –

  Months to run in (e.g. ["jan", "apr", "jul", "oct"]).

• every_month

  (int | None, default: None ) –

  Run every N months.

• timezone

  (str | None, default: None ) –

  IANA timezone name (e.g. "America/New_York").

• run_after_job_uuid

  (str | None, default: None ) –

  UUID of a job that must complete before this schedule triggers.

• run_after_job_name

  (str | None, default: None ) –

  Name of a job that must complete before this schedule triggers.

• run_after_job_condition

  (str | None, default: None ) –

  Required completion status of the predecessor job (e.g. "success").

Example

from datatailr import Schedule
# Every weekday at 08:00 and 16:00 UTC
s = Schedule(at_hours=[8, 16], weekdays=["mon","tue","wed","thu","fri"])

get_cron_string

Return the compiled cron string.

Returns:

• str –

  Cron string (str).

Examples:

>>> Schedule(
...     at_minutes=[0, 15, 30, 45],
...     at_hours=[0, 12],
...     weekdays=["Mon", "Wed", "Fri"],
...     day_of_month=15,
...     in_month=["Jan", "Jul"],
... ).get_cron_string()
'0 0,15,30,45 0,12 15 1,7 1,3,5'