Cron

class chancy.plugins.cron.Cron(*, poll_interval: int = 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.

Note

While the underlying library used to parse the cron syntax supports timezones, this plugin does not. All times are assumed to be in UTC. This is due to frequent issues that occur with timezones and daylight saving time changes that we simply don’t want to support.

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.pruner import Cron
from chancy.plugins.leadership import Leadership

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

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

param poll_interval:

The number of seconds between cron poll intervals.

api_plugin() → str | None

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

classmethod get_scope() → PluginScope

Get the scope of this plugin. Scopes control when and where the plugin will be run.

By default, plugins are scoped to the worker.

migrate_key() → str | None

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

migrate_package() → str | None

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

async run(worker: Worker, chancy: Chancy)

Runs the plugin.

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

async classmethod schedule(chancy: Chancy, cron: str, *jobs: Job | IsAJob)

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.

Parameters:

• chancy – The Chancy application.

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

• jobs – The jobs to run.

async classmethod unschedule(chancy: Chancy, *unique_keys: str)

Permanently unschedule one or more jobs from running.

await Cron.unschedule(chancy, "hello_world_cron")

Parameters:

• chancy – The Chancy application.

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

class chancy.plugins.cron.api.CronApiPlugin

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.