API reference

Everything public is importable from the package root:

from action0.celery_sched import (
    BeatEntry,
    DefinitionError,
    DuplicateEntryError,
    Entry,
    Format,
    FormatLike,
    ScheduleError,
    SolarEvent,
    Source,
    UnknownTaskError,
    check_tasks,
    load_beat_schedule,
    load_entries,
    parse_crontab,
    parse_duration,
    parse_entry,
    parse_schedule,
    parse_solar,
)

Loading

Read schedules and turn them into Celery’s beat_schedule setting.

A schedule is a mapping of entry names to entries (see entries), written in YAML or TOML (see formats) or handed over as a mapping (see sources); an empty file is an empty schedule. Entries whose name starts with a . are templates: they are never scheduled. In YAML, that makes them the place for anchors that other entries merge in:

.reports: &reports
  task: myapp.reports.tasks.build
  options: {queue: reports}

"Daily report":
  <<: *reports
  kw: {period: day}
  schedule: {crontab: "@daily"}

TOML cannot refer to another table, so there templates are merely skipped.

action0.celery_sched.loader.load_beat_schedule(*sources, replace=False, format=None)[source]

Read schedules into a value for Celery’s beat_schedule setting.

Entries with enabled: false are left out:

app.conf.beat_schedule = load_beat_schedule("beat.yaml")  # or "beat.toml"

>>> load_beat_schedule({"Poll feed": {"task": "myapp.tasks.poll", "schedule": {"every": "5m"}}})
{'Poll feed': {'task': 'myapp.tasks.poll', 'schedule': <freq: 5.00 minutes>, 'args': (), 'kwargs': {}, 'options': {}}}
Parameters:
Return type:

dict[str, BeatEntry]

Returns:

entry name → beat entry, in file order

Raises:
action0.celery_sched.loader.load_entries(*sources, replace=False, format=None)[source]

Read and validate the entries of one or more schedules.

Unlike load_beat_schedule(), disabled entries are included, and each one knows the file it came from — handy for tooling and tests.

Sources are merged in order, whatever their format. An entry name defined twice is an error; with replace=True the later definition wins instead (keeping the position of the first), so an environment-specific file can override or disable entries of a shared one.

Parameters:

  • sources (str | PathLike[str] | IO[str] | IO[bytes] | Mapping[str, Any]) – the files, streams or mappings to read

  • replace (bool, default: False) – let a later source redefine an entry instead of raising

  • format (Format | Literal['yaml', 'toml'] | None, default: None) – the format of every file and stream, instead of telling it from their names

Return type:

list[Entry]

Returns:

the entries, in file order

Raises:

Sources

What the loaders read from, and turning each kind of source into data.

A source is one of:

  • a path (a str or any path-like) to a YAML or TOML file;

  • an open stream — text or binary, as tomllib users are used to;

  • a mapping that is already the parsed document, e.g. a table of a larger TOML configuration, or a dict from a settings system.

action0.celery_sched.sources.Source: TypeAlias = str | os.PathLike[str] | typing.IO[str] | typing.IO[bytes] | collections.abc.Mapping[str, typing.Any]

what the load functions read from: a file path, an open stream, or parsed data

action0.celery_sched.sources.read_source(source, format=None)[source]

Read a source into its parsed document, and name it for error messages.

>>> read_source({"Poll": {"task": "myapp.tasks.poll"}})
(None, {'Poll': {'task': 'myapp.tasks.poll'}})
Parameters:

  • source (str | PathLike[str] | IO[str] | IO[bytes] | Mapping[str, Any]) – the source to read

  • format (Format | Literal['yaml', 'toml'] | None, default: None) – the format of a file or stream, instead of guessing it from the name (ignored for mappings, which are parsed already)

Return type:

tuple[str | None, Any]

Returns:

the source’s name (the path, the stream’s name, or None) and its document

Raises:

  • ValueError – if the format of a file or stream can’t be told

  • ImportError – for a YAML source, if PyYAML (the yaml extra) is not installed

  • DefinitionError – if the text is malformed

Formats

The file formats a schedule can be written in, and how the loaders tell them apart.

A schedule means the same whatever its format — the entries have the same keys and values — so the format only decides which parser turns the text into data. By default a file’s suffix decides: .yaml and .yml are YAML, .toml is TOML. A stream is judged by its name (which files opened with open() have); anything else needs an explicit format.

TOML is read with the standard library’s tomllib. YAML needs PyYAML, which comes with the yaml extra (pip install "action0-celery-sched[yaml]"); it is imported only once a YAML source is actually read, so TOML-only installs never need it.

class action0.celery_sched.formats.Format(*values)[source]

A file format for schedules.

>>> Format("toml") is Format.TOML
True
YAML = 'yaml'
TOML = 'toml'
action0.celery_sched.formats.FormatLike: TypeAlias = action0.celery_sched.formats.Format | typing.Literal['yaml', 'toml']

a Format, or its value as a plain string

action0.celery_sched.formats.SUFFIXES: dict[str, Format] = {'.toml': Format.TOML, '.yaml': Format.YAML, '.yml': Format.YAML}

the file suffixes each format is recognised by (compared case-insensitively)

action0.celery_sched.formats.detect_format(name, format=None)[source]

Decide the format of a file: the explicit one if given, else by its suffix.

>>> detect_format("conf/beat.toml")
<Format.TOML: 'toml'>
>>> detect_format(None, "yaml")
<Format.YAML: 'yaml'>
Parameters:

  • name (str | None) – the file name or path, None for an unnamed stream

  • format (Format | Literal['yaml', 'toml'] | None, default: None) – the format to use regardless of the name

Return type:

Format

Returns:

the format

Raises:

ValueError – if format is not a known format, or it isn’t given and the name doesn’t end in a known suffix

action0.celery_sched.formats.parse_text(text, format)[source]

Parse schedule text in the given format.

>>> parse_text("[Poll]\ntask = 'myapp.tasks.poll'", Format.TOML)
{'Poll': {'task': 'myapp.tasks.poll'}}
Parameters:

  • text (str) – the document

  • format (Format) – its format

Return type:

Any

Returns:

the parsed document, as plain Python data

Raises:
action0.celery_sched.formats.parser_for(format)[source]

Get the parser of a format, importing the YAML one on first use.

The loaders call this before reading a source, so a missing PyYAML is reported without any I/O, like a format that can’t be told.

>>> parser_for(Format.TOML).__name__
'load_toml'
Parameters:

format (Format) – the format

Return type:

Callable[[str], Any]

Returns:

a function turning text in that format into plain Python data

Raises:

ImportError – for YAML, if PyYAML (the yaml extra) is not installed

Entries

One schedule entry: its shape in a file, its parsed form and Celery’s.

An entry is a named mapping, with the same keys in YAML and TOML:

"Nightly report":                     # the entry's name, unique
  task: myapp.reports.tasks.nightly   # the registered task name — required
  schedule:                           # when to run it — required
    crontab: "0 3 * * *"
  params: [daily]                     # positional arguments
  kw:                                 # keyword arguments
    recipients: [ops@example.com]
  options:                            # passed on to apply_async()
    queue: reports
  enabled: true                       # false keeps the entry but skips it

["Nightly report"]                          # the entry's name, unique
task = "myapp.reports.tasks.nightly"        # the registered task name — required
schedule = { crontab = "0 3 * * *" }        # when to run it — required
params = ["daily"]                          # positional arguments
kw = { recipients = ["ops@example.com"] }   # keyword arguments
options = { queue = "reports" }             # passed on to apply_async()
enabled = true                              # false keeps the entry but skips it

Any other key is an error, so a typo like shedule fails at startup instead of silently dropping the schedule.

action0.celery_sched.entries.ENTRY_KEYS = ('task', 'schedule', 'params', 'kw', 'options', 'enabled')

every key an entry may have

class action0.celery_sched.entries.BeatEntry[source]

One value of Celery’s beat_schedule setting, as beat expects it.

task: str
schedule: BaseSchedule
args: tuple[Any, ...]
kwargs: dict[str, Any]
options: dict[str, Any]
class action0.celery_sched.entries.Entry(name, task, schedule, args=(), kwargs=<factory>, options=<factory>, enabled=True, source=None)[source]

A parsed and validated schedule entry.

>>> entry = parse_entry("Poll feed", {"task": "myapp.tasks.poll", "schedule": {"every": 30}})
>>> entry.to_celery()
{'task': 'myapp.tasks.poll', 'schedule': <freq: 30.00 seconds>, 'args': (), 'kwargs': {}, 'options': {}}
Parameters:
name: str

the entry’s name — the key in beat_schedule

task: str

the name the task is registered under

schedule: BaseSchedule

when the task runs

args: tuple[Any, ...] = ()

positional arguments (params in the file)

kwargs: dict[str, Any]

keyword arguments (kw in the file)

options: dict[str, Any]

apply_async() options such as queue, priority or expires

enabled: bool = True

whether the entry goes into the beat schedule at all

source: str | None = None

the file (or stream name) the entry was read from, if known

to_celery()[source]

Render the entry as a beat_schedule value.

Return type:

BeatEntry

Returns:

a fresh dict; mutating it does not affect the entry

action0.celery_sched.entries.parse_entry(name, value, *, source=None)[source]

Validate one entry of a schedule file.

The argument containers are deep-copied, so entries sharing a mapping through a YAML anchor never share it at runtime.

Parameters:

  • name (str) – the entry’s name (its key in the file)

  • value (object) – the parsed value of the entry, from either format

  • source (str | None, default: None) – where the entry was read from, recorded on the entry

Return type:

Entry

Returns:

the entry

Raises:

DefinitionError – if the entry is malformed

Schedules

Parse the schedule of an entry.

It is a mapping with exactly one key naming the kind of schedule: every (an interval, see durations), crontab (see crontabs) or solar (see solar). The same three entries in each format:

"Poll":   {task: myapp.tasks.poll, schedule: {every: 1h, relative: true}}
"Report": {task: myapp.tasks.report, schedule: {crontab: "0 3 * * *"}}
"Lights": {task: myapp.tasks.lights, schedule: {solar: {event: sunset, lat: 48.21, lon: 16.37}}}

Poll = { task = "myapp.tasks.poll", schedule = { every = "1h", relative = true } }
Report = { task = "myapp.tasks.report", schedule = { crontab = "0 3 * * *" } }
Lights = { task = "myapp.tasks.lights", schedule = { solar = { event = "sunset", lat = 48.21, lon = 16.37 } } }

relative is Celery’s own flag for intervals: when true, the time of the next run is rounded to the resolution of the interval (a 1h interval runs on the hour). It is only allowed next to every.

action0.celery_sched.schedules.KINDS = ('every', 'crontab', 'solar')

the keys naming a kind of schedule; exactly one of them must be given

action0.celery_sched.schedules.parse_schedule(value)[source]

Turn the schedule of an entry into a Celery schedule.

Intervals become a celery.schedules.schedule, so every kind comes back as a BaseSchedule that beat accepts as it is.

>>> parse_schedule({"every": "5m"})
<freq: 5.00 minutes>
>>> parse_schedule({"crontab": "0 3 * * *"})
<crontab: 0 3 * * * (m/h/dM/MY/d)>
Parameters:

value (object) – the parsed value, from either format

Return type:

BaseSchedule

Returns:

the schedule

Raises:

DefinitionError – if the block is malformed

Intervals

Parse the interval of an every schedule.

Three spellings are accepted:

  • a number of seconds: 30, 2.5 (or the same as a string, which is what !ENV produces);

  • a duration string of <amount><unit> parts: 90s, 5m, 1h30m, 1.5h, 2d 12h — units w, d, h, m, s and ms, lowercase only (so m can never be misread as months);

  • a mapping of timedelta arguments: {hours: 1, minutes: 30} in YAML, { hours = 1, minutes = 30 } in TOML.

action0.celery_sched.durations.parse_duration(value)[source]

Turn the value of an every key into a positive timedelta.

>>> parse_duration(90)
datetime.timedelta(seconds=90)
>>> parse_duration("1h30m")
datetime.timedelta(seconds=5400)
>>> parse_duration({"days": 1, "hours": 12})
datetime.timedelta(days=1, seconds=43200)
Parameters:

value (object) – the parsed value, from either format

Return type:

timedelta

Returns:

the interval

Raises:

DefinitionError – if the value is malformed, zero or negative

Crontabs

Parse the value of a crontab schedule into a celery.schedules.crontab.

Two spellings are accepted:

  • a classic five-field cron string, "minute hour day-of-month month day-of-week" — for example "30 7 * * mon-fri" — or one of the nicknames @hourly, @daily/@midnight, @weekly, @monthly and @yearly/@annually;

  • a mapping of Celery’s own keyword arguments, where omitted fields mean *: {minute: 30, hour: 7, day_of_week: mon-fri} in YAML, { minute = 30, hour = 7, day_of_week = "mon-fri" } in TOML.

Each field takes whatever Celery takes — */15, 1-5, mon,wed, a plain number — and in the mapping form also a list of numbers. Every field is validated by Celery while loading, so a typo fails at startup, not when beat first evaluates the schedule.

action0.celery_sched.crontabs.parse_crontab(value)[source]

Turn the value of a crontab key into a Celery crontab.

>>> parse_crontab("30 7 * * mon-fri")
<crontab: 30 7 * * mon-fri (m/h/dM/MY/d)>
>>> parse_crontab({"minute": "*/15"})
<crontab: */15 * * * * (m/h/dM/MY/d)>
>>> parse_crontab("@daily")
<crontab: 0 0 * * * (m/h/dM/MY/d)>
Parameters:

value (object) – the parsed value, from either format

Return type:

crontab

Returns:

the crontab

Raises:

DefinitionError – if the value is malformed or a field is invalid

Solar events

Parse the value of a solar schedule into a celery.schedules.solar.

The value is a mapping of the event and the observer’s position: solar: {event: sunset, lat: 48.21, lon: 16.37} in YAML, solar = { event = "sunset", lat = 48.21, lon = 16.37 } in TOML.

Celery computes the event times with ephem, which is only installed with the solar extra (pip install "action0-celery-sched[solar]").

class action0.celery_sched.solar.SolarEvent(*values)[source]

The events a solar schedule can fire on, as Celery names them.

>>> SolarEvent("sunset") is SolarEvent.SUNSET
True
DAWN_ASTRONOMICAL = 'dawn_astronomical'
DAWN_NAUTICAL = 'dawn_nautical'
DAWN_CIVIL = 'dawn_civil'
SUNRISE = 'sunrise'
SOLAR_NOON = 'solar_noon'
SUNSET = 'sunset'
DUSK_CIVIL = 'dusk_civil'
DUSK_NAUTICAL = 'dusk_nautical'
DUSK_ASTRONOMICAL = 'dusk_astronomical'
action0.celery_sched.solar.parse_solar(value)[source]

Turn the value of a solar key into a Celery solar schedule.

>>> parse_solar({"event": "sunset", "lat": 48.21, "lon": 16.37})
<solar: sunset at latitude 48.21, longitude: 16.37>
Parameters:

value (object) – the parsed value, from either format

Return type:

solar

Returns:

the solar schedule

Raises:

Task check

Check that every scheduled task is actually registered with the Celery app.

Beat sends tasks by name, and a misspelled name is not noticed until the first run is due — possibly hours after a deploy — and then only as an error on the worker. check_tasks() catches it at startup.

The app only knows the tasks whose modules were imported. Beat imports them (imports/include and autodiscovery) right before it sends the beat_init signal, which makes that the place to check:

from celery.signals import beat_init


@beat_init.connect
def check_schedule(sender, **kwargs):
    check_tasks(sender.app)
action0.celery_sched.tasks.check_tasks(app, schedule=None)[source]

Raise if a schedule entry refers to a task the app hasn’t registered.

>>> app = Celery("example")
>>> @app.task(name="myapp.tasks.poll")
... def poll() -> None: ...
>>> check_tasks(app, {"Poll feed": {"task": "myapp.tasks.poll"}})
>>> check_tasks(app, {"Report": {"task": "myapp.tasks.report"}})
Traceback (most recent call last):
...
action0.celery_sched.errors.UnknownTaskError: unregistered tasks: 'Report' -> myapp.tasks.report
Parameters:

  • app (Celery) – the Celery app whose task registry to check against

  • schedule (Mapping[str, Mapping[str, Any]] | None, default: None) – the entries to check, in beat_schedule shape (default: the app’s own beat_schedule setting, so entries configured some other way are checked too)

Raises:

UnknownTaskError – listing every entry whose task is unregistered

Return type:

None

Environment variables

The !ENV substitution: ${VAR} and ${VAR:-fallback} from the environment.

Substitution is purely textual and happens once, at load time: the result is always a string and is not parsed as YAML again, so a value like "yes" or "0123" cannot change type behind your back. The validators accept the string spellings of numbers and booleans for exactly that reason.

action0.celery_sched.envvars.substitute_env(value, environ=None)[source]

Replace ${VAR} / ${VAR:-fallback} occurrences from the environment.

>>> substitute_env("${REPORT_HOUR:-3}", environ={})
'3'
>>> substitute_env("${QUEUE}-high", environ={"QUEUE": "reports"})
'reports-high'
Parameters:

  • value (str) – the scalar tagged with !ENV

  • environ (Mapping[str, str] | None, default: None) – where to look the variables up (default: os.environ)

Return type:

str

Returns:

the substituted string

Raises:

DefinitionError – if a variable without fallback is not set

YAML loader

The YAML loader for schedule files: PyYAML’s SafeLoader, plus two things.

  • The !ENV tag substitutes environment variables into a scalar (see envvars).

  • Duplicate keys are an error. Plain YAML parsers silently keep the last of two equal keys, so a copy-pasted entry that kept its name would quietly replace the original. Keys brought in through a merge (<<: *base) may still be overridden — that is what merging is for.

Being a SafeLoader, it cannot instantiate arbitrary Python objects. load_yaml() is the entry point the loaders use.

class action0.celery_sched.yaml_loader.ScheduleLoader(stream)[source]

SafeLoader with the !ENV tag and duplicate-key detection.

>>> yaml.load("a: 1\na: 2", Loader=ScheduleLoader)
Traceback (most recent call last):
...
action0.celery_sched.errors.DuplicateEntryError: duplicate entry 'a' on line 2 (first on line 1)
Parameters:

  • stream (Any)

  • stream – the YAML text or stream

construct_document(node)[source]

Remember the root, so a duplicate entry can be told from a duplicate key.

Parameters:

node (Node)

Return type:

Any

flatten_mapping(node)[source]

Check a mapping’s own keys for duplicates, then resolve its merges.

This is the one hook PyYAML calls on every mapping before it rewrites the node’s pairs to include the merged ones. A node can be flattened more than once (as a merge source, then again when constructed), so each is checked only on its first, untouched visit.

Parameters:

node (MappingNode)

Return type:

None

action0.celery_sched.yaml_loader.load_yaml(text)[source]

Parse YAML text with the ScheduleLoader.

>>> load_yaml("'Poll feed': {schedule: {every: 5m}}")
{'Poll feed': {'schedule': {'every': '5m'}}}
Parameters:

text (str) – the YAML document

Return type:

Any

Returns:

the parsed document (None for an empty one)

Raises:

DefinitionError – on a syntax error, a duplicate key, or an unset environment variable

TOML loader

Parse TOML schedule files, with the standard library’s tomllib.

An entry in TOML has exactly the keys and values it has in YAML:

["Nightly report"]
task = "myapp.reports.tasks.nightly"
params = ["daily"]
kw = { recipients = ["ops@example.com"] }
schedule = { crontab = "0 3 * * *" }
options = { queue = "reports" }

TOML has no tags, so the !ENV substitution is written as a prefix of a string: crontab = "!ENV ${REPORT_CRON:-0 3 * * *}". The rest of the string, after !ENV and the whitespace following it, is substituted exactly like YAML’s !ENV tag does it (see envvars). Only string values are substituted, never keys.

TOML itself forbids declaring a key twice, so a duplicate entry is a syntax error here.

action0.celery_sched.toml_loader.load_toml(text)[source]

Parse TOML text and substitute its !ENV-prefixed strings.

>>> load_toml('["Poll feed"]\nschedule = { every = "!ENV ${POLL_EVERY:-5m}" }')
{'Poll feed': {'schedule': {'every': '5m'}}}
Parameters:

text (str) – the TOML document

Return type:

dict[str, Any]

Returns:

the parsed document

Raises:

DefinitionError – on a syntax error, or an unset environment variable

Errors

Exceptions raised while reading schedules; everything derives from ScheduleError.

exception action0.celery_sched.errors.ScheduleError[source]

Base class of every error this package raises on purpose.

exception action0.celery_sched.errors.DefinitionError(reason)[source]

A schedule file, or one entry in it, is malformed.

The error knows where the problem is: the file it came from, the entry and the key path inside that entry. The parsers only state the reason; the location is filled in on the way up (see located()), so the message reads like:

beat.yaml: entry 'Nightly report': schedule.crontab: expected 5 fields, got 4

>>> error = DefinitionError("expected 5 fields, got 4")
>>> error.source, error.entry, error.path = "beat.yaml", "Nightly report", ("schedule", "crontab")
>>> print(error)
beat.yaml: entry 'Nightly report': schedule.crontab: expected 5 fields, got 4
Parameters:

  • reason (str)

  • reason – what is wrong, without the location

source: str | None

the file (or stream name) the entry was read from, if known

entry: str | None

the name of the entry, if the error is inside one

path: tuple[str, ...]

the keys leading from the entry to the offending value

exception action0.celery_sched.errors.DuplicateEntryError(reason)[source]

Two entries share a name — within one file, or across files merged without replace=True.

Parameters:

  • reason (str)

  • reason – what is wrong, without the location

exception action0.celery_sched.errors.UnknownTaskError(missing)[source]

Entries refer to tasks the Celery app has not registered.

Raised by check_tasks(), listing every offender at once rather than only the first.

>>> print(UnknownTaskError({"Nightly report": "myapp.tasks.nightly"}))
unregistered tasks: 'Nightly report' -> myapp.tasks.nightly
Parameters:

  • missing (Mapping[str, str])

  • missing – entry name → the task name nothing is registered under

action0.celery_sched.errors.located(*keys, entry=None, source=None)[source]

Add location details to any DefinitionError raised inside the block.

Nest these as the parsers descend into a value, so each layer only needs to know its own key (or entry name, or file):

>>> try:
...     with located(entry="Poll feed"):
...         with located("schedule"):
...             with located("every"):
...                 raise DefinitionError("must be positive")
... except DefinitionError as error:
...     print(error)
entry 'Poll feed': schedule.every: must be positive
Parameters:

  • keys (str) – the keys this block descends into, prepended to the path

  • entry (str | None, default: None) – the entry this block parses, unless the error already has one

  • source (str | None, default: None) – the file this block reads, unless the error already has one

Return type:

Iterator[None]