Skip to content

API

pico_scheduling.decorators

Marker decorator: stores the schedule on the method; the registrar picks it up at container startup (same idiom as pico-celery's @task).

scheduled(*, every=None, cron=None)

Run a component method on a schedule.

Exactly one of every (seconds between runs) or cron (standard 5-field crontab expression) must be given. Sync and async methods are both supported.

Source code in src/pico_scheduling/decorators.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def scheduled(*, every: float | None = None, cron: str | None = None):
    """Run a component method on a schedule.

    Exactly one of ``every`` (seconds between runs) or ``cron`` (standard
    5-field crontab expression) must be given. Sync and async methods are
    both supported.
    """
    if (every is None) == (cron is None):
        raise ValueError("@scheduled requires exactly one of every= or cron=")
    if every is not None and every <= 0:
        raise ValueError("@scheduled(every=...) requires a positive number of seconds")

    def dec(fn):
        setattr(fn, SCHEDULED_META, {"every": every, "cron": cron})
        return fn

    return dec

pico_scheduling.registrar

Discovers @scheduled methods and runs them on a background scheduler.

Jobs resolve their component through the container on every run, so prototype-scoped components get a fresh instance per execution and singletons are shared — same semantics as pico-celery task dispatch. Async methods run via asyncio.run inside the scheduler's worker thread.

pico_scheduling.config

Settings for pico-scheduling (prefix scheduling, zero-config).

SchedulingSettings dataclass

enabled: false disables every @scheduled job.

Source code in src/pico_scheduling/config.py 
 8
 9
10
11
12
13
@configured(target="self", prefix="scheduling", mapping="tree")
@dataclass
class SchedulingSettings:
    """``enabled: false`` disables every ``@scheduled`` job."""

    enabled: bool = True