Cron

class chancy.plugins.cron.Cron(*, poll_interval=60)

Bases: Plugin

Run jobs at specific times and intervals using cron-like syntax.

The schedule is persistent and dynamic, being stored in the database. This allows for jobs to be scheduled and rescheduled without needing to restart the worker(s).

If a scheduled job is already on the queue waiting to run, or currently running, the job will not be queued again and instead will wait until the next scheduled time.

Timezones

By default, cron expressions are evaluated in UTC. Each schedule can instead be evaluated in its own IANA timezone by passing timezone to schedule():

from zoneinfo import ZoneInfo

# Every day at 09:00 Paris time, whether it is winter or summer.
await Cron.schedule(
    chancy,
    "0 9 * * *",
    hello_world.job.with_unique_key("hello_world_paris"),
    timezone=ZoneInfo("Europe/Paris"),
)

Daylight saving time

UTC, the default, and timezones without daylight saving time (DST), such as Asia/Tokyo, are never affected by this section.

In a timezone that observes DST, the local clock skips a period once a year and repeats one once a year. How long that period is, often one hour, and when it happens depend on the timezone. In Europe/Paris, 02:00-03:00 does not exist on the last Sunday of March and happens twice on the last Sunday of October (first at UTC+2, then at UTC+1). In Australia/Lord_Howe the period is 30 minutes, and in America/Santiago the change happens at midnight. Schedules outside the skipped and repeated periods are unaffected and keep their local time all year round.

Inside those periods, the behaviour depends on whether the job runs at a fixed time or uses a wildcard:

• A fixed-time job has neither its minute nor its hour field starting with *, e.g. 30 2 * * *, 0 2 * * 1-5 or @daily.

• A wildcard job has its minute or its hour field starting with *, e.g. * * * * *, */15 2 * * *, 0 * * * *, 0 */2 * * * or @hourly.

These are the classes, and the rules below, that Vixie cron and its descendants such as Debian’s cron use for daylight saving time changes (see cron(8)). Like cron, the class depends on how the fields are written, not on the times they match. 0 */2 * * * is a wildcard job, but the equivalent 0 0/2 * * * is a fixed-time job, and so is 0 0-23 * * * while 0 * * * * is a wildcard job.

Transition (example: Europe/Paris)	Fixed-time job (30 2 * * *)	Wildcard job (*/30 * * * *)
Skipped period (02:00-03:00 in spring)	Runs once, at the end of the skipped period (03:00), then at 02:30 again from the next day.	Runs at the times that exist: 01:30, then 03:00. The skipped 02:00 and 02:30 do not run. A job restricted to the skipped period, like */15 2 * * *, does not run that day.
Repeated period (02:00-03:00 in autumn)	Runs once, during the first pass (02:30 UTC+2). The second 02:30 (UTC+1) does not run.	Matches local times in both passes: 02:00 and 02:30 UTC+2, then 02:00 and 02:30 UTC+1.

A fixed-time job does not repeat an occurrence because of DST. Wildcard jobs match the local clock, so elapsed intervals can change across DST. For example, 15 */2 * * * runs at 00:15 and then 04:15 when 02:15 is skipped; it does not catch up at 03:00. tests/plugins/test_cron_dst.py covers these rules for Europe/Paris, Australia/Lord_Howe (30-minute shift), Antarctica/Troll (2-hour shift) and America/Santiago (change at midnight).

Limitations

• Some wildcard schedules can miss occurrences during the repeated DST period due to a croniter limitation. For example, */7 * * * * can skip the second pass in Australia/Lord_Howe.

• A schedule must have a usable occurrence within the next 50 calendar years. Schedules without one are rejected when saved and skipped with an error in the logs when already stored.

• next_run is computed when a schedule is saved through schedule() and each time it runs. Editing the expression or the timezone directly in the database, including through the Django admin, only takes effect after the next run.

• The timezone column must hold a valid IANA name, and the expression must be valid. The Django admin only offers valid timezones, but does not validate expressions. A due schedule with an invalid timezone or expression is skipped, logging an error on every poll until it is fixed.

• Versions of Chancy without timezone support ignore the column and evaluate every schedule in UTC. While workers of both versions run side by side, or after a downgrade, schedules in another timezone can fire at the UTC reading of their expression.

• The dashboard’s timeline computes upcoming runs in the browser. It uses each schedule’s timezone, but around a DST transition it can differ from the actual runs: it may omit executions or show extra or shifted ones. The Next Run value comes from the server’s stored schedule.

Installation

This plugin requires an extra dependency to parse the cron syntax. You can install it using:

pip install chancy[cron]

This plugin requires a database migration to create the table that stores the cron-like schedules.

Usage

To use the cron plugin, you need to add it to your Chancy application and then set up the schedule for the jobs you want to run:

import asyncio
from chancy import Chancy, Worker, Queue, job
from chancy.plugins.cron import Cron

@job(queue="default")
def hello_world():
    print("hello_world")

async with Chancy(
    "postgresql://localhost/postgres",
    plugins=[Cron()]
) as chancy:
    await Cron.schedule(
        chancy,
        "*/2 * * * *",
        hello_world.job.with_unique_key("hello_world_cron")
    )

Django Integration

This plugin can be made available to the Django ORM and Admin interface.

To enable this, you need to add the following to your Django settings:

INSTALLED_APPS = [
    ...,
    "chancy.plugins.cron.django",
]

You can then query the scheduled jobs using the Django ORM:

from chancy.plugins.cron.django.models import Cron

# Get all scheduled jobs
all_schedules = Cron.objects.all()

param poll_interval:

The number of seconds between cron poll intervals.

api_plugin()

If this plugin has an associated API component, returns the import string for the plugin.

Return type:

str | None

static get_identifier()

Returns a unique identifier for this plugin.

This identifier should be unique across all active plugins. If a custom plugin provides compatible functionality to a built-in plugin, it may use the same identifier as the built-in plugin.

Return type:

str

async classmethod get_schedules(chancy, *, unique_keys=None)

Get scheduled cron jobs by their unique keys.

If no unique keys are provided, all scheduled jobs will be returned.

# Get all scheduled jobs
all_schedules = await Cron.get_schedules(chancy)

# Get a specific job
job_schedule = await Cron.get_schedules(chancy, ["hello_world_cron"])

Parameters:

• chancy (Chancy) – The Chancy application.

• unique_keys (list[str] | None) – Optional list of unique keys to filter by.

Returns:

The schedules, keyed by unique key. Each one is a dictionary with unique_key, job, cron, timezone (its IANA name), last_run and next_run.

Return type:

dict[str, dict]

get_tables()

Get the names of all tables this plugin is responsible for.

Return type:

list[str]

migrate_key()

Get the migration key for this plugin, if it has any.

Return type:

str | None

migrate_package()

Get the package name that contains the migration scripts for this plugin, if it has any.

Return type:

str | None

async run(worker, chancy)

Runs the plugin.

This function can and should run indefinitely, as it will be cancelled when the worker is stopped.

Parameters:

• worker (Worker)

• chancy (Chancy)

async classmethod schedule(chancy, cron, *jobs, timezone=zoneinfo.ZoneInfo(key='Etc/UTC'))

Schedule one or more jobs to run periodically based on a cron schedule.

All jobs that are scheduled with this feature must be using a unique_key to ensure that only one copy of the job is scheduled at a time. Scheduling a job with the same unique key as an existing job will update the existing job with the new schedule, job & timezone.

Parameters:

• chancy (Chancy) – The Chancy application.

• cron (str) – A cron-like syntax string that describes when to run the job.

• jobs (Job | IsAJob) – The jobs to run.

• timezone (ZoneInfo) – The timezone in which to evaluate the cron expression. Defaults to UTC.

async classmethod unschedule(chancy, *unique_keys)

Permanently unschedule one or more jobs from running.

await Cron.unschedule(chancy, "hello_world_cron")

Parameters:

• chancy (Chancy) – The Chancy application.

• unique_keys (str) – The unique keys of the jobs to unschedule.

Parameters:

poll_interval (int)

class chancy.plugins.cron.api.CronApiPlugin(api)

Bases: ApiPlugin

Provides API endpoints for viewing cron jobs.

async static get_cron(request, *, chancy, worker)

Get all known cron jobs.

name()

Get the name of the plugin.

routes()

Get a list of routes to add to the API.