aiocron Documentation

The simplest way to schedule an asynchronous function is by using the @aiocron.crontab decorator. This will automatically schedule the decorated function to run according to the provided cron expression when the asyncio event loop is running.

Note: aiocron uses cronsim for parsing cron expressions. Ensure your expressions are compatible with cronsim.

import aiocron
import asyncio

@aiocron.crontab('*/30 * * * *')
async def attime():
    print('run')

asyncio.get_event_loop().run_forever()

You can use @aiocron.crontab with start=False to prevent the task from starting immediately. This turns the decorated function into a crontab object that you can control manually.

• Use .start() to begin the scheduled execution.
• Access the original decorated function via the .func attribute.
• Use .next(*args) to await the next scheduled execution, which allows you to pass arguments to the coroutine.

import aiocron
import asyncio

# Schedule but don't start immediately
@aiocron.crontab('1 9 * * 1-5', start=False)
async def attime():
    print('run')

# Start the schedule manually
attime.start()

# Access the original function
original_func = attime.func

asyncio.get_event_loop().run_forever()

You can use crontab(...).next() as a way to sleep until the next time a specific cron schedule occurs. This is useful for delaying execution until a specific time window.

import aiocron
import asyncio

async def main():
    # Wait until the next hour
    await aiocron.crontab('0 * * * *').next()
    print("It is now the start of the hour")

asyncio.run(main())

If you prefer not to use decorators, you can instantiate a crontab object manually by passing your coroutine to the func parameter of crontab().

import aiocron

async def yourcoroutine():
    print('running')

# Manual instantiation
cron = aiocron.crontab('0 * * * *', func=yourcoroutine, start=False)

# You can then control it via cron.start(), etc.
cron.start()

If you need to wait for the next occurrence of a scheduled task, you can await the .next() method of a crontab object. This method accepts arguments that will be passed to the decorated coroutine when it executes.

@aiocron.crontab('0 9,10 * * * mon,fri', start=False)
async def attime(i):
    print('run %i' % i)

# In another coroutine:
res = await attime.next(1) # Passes 1 as argument 'i'

import aiocron
import asyncio

@aiocron.crontab('0 9,10 * * * mon,fri', start=False)
async def attime(i):
    print('run %i' % i)

async def once():
    try:
        # Await the next execution and pass an argument
        res = await attime.next(1)
    except Exception as e:
        print('It failed (%r)' % e)
    else:
        print(res)

asyncio.run(once())

When instantiating Cron or using crontab, you can provide specific configuration for execution:

• args: A tuple of positional arguments passed to the function.
• kwargs: A dictionary of keyword arguments passed to the function.
• tz: A timezone object (e.g., from pytz or zoneinfo) to ensure the cron schedule respects local time.
• loop: An explicit asyncio event loop instance.

import aiocron
from datetime import datetime

# Example with args, kwargs, and a specific timezone
job = aiocron.crontab(
    '*/1 * * * *',
    args=(1, 2),
    kwargs={'key': 'value'},
    tz='America/New_York'
)

The crontab function can be used as a decorator to wrap an asynchronous or synchronous function, scheduling it to run according to a cron specification. By default, start=True, meaning the scheduler begins immediately upon decoration.

Arguments:

• spec: The cron specification string.
• func: The function to schedule (if not using as a decorator).
• args: Positional arguments for the function.
• kwargs: Keyword arguments for the function.
• start: Whether to start the scheduler immediately (default True).
• loop: The asyncio event loop to use.
• tz: The timezone to use for scheduling.

import aiocron

@aiocron.crontab('*/5 * * * *')
async def my_task():
    print("Running every 5 minutes")

The Cron class provides a low-level interface for managing scheduled tasks. You can instantiate it with a cron spec and a function, then manually control its lifecycle using .start() and .stop().

Key methods:

• start(): Begins the scheduling loop.
• stop(): Cancels the current scheduled handle and stops the loop.
• next(*args): An awaitable that waits until the next scheduled execution occurs, then runs the function with the provided *args.

import asyncio
import aiocron

async def main():
    # Initialize Cron manually
    job = aiocron.Cron('0 * * * *', my_coroutine, start=False)
    
    # Start the job
    job.start()
    
    # Wait for the next execution
    await job.next()
    print("The job just ran!")

async def my_coroutine():
    print("Task executed")

asyncio.run(main())

You can use aiocron as a standalone CLI tool to execute shell commands on a cron schedule using python -m aiocron. This is useful for running simple periodic tasks without writing a custom Python script.

Arguments:

• crontab: A quoted cron expression (e.g., "* * * * *").
• command: One or more shell commands to execute.

Options:

• -n <int>: The number of times to loop the command. Set to 0 for an infinite loop (default is 1).

If an invalid cron format is provided, the CLI will exit with an error.

# Run a command every minute, infinite loop
python -m aiocron "* * * * *" echo "hello world"

# Run a command 5 times based on a cron schedule
python -m aiocron -n 5 "*/5 * * * *" python my_script.py