celery-sqlalchemy-v2-scheduler 0.1.2

Celery beat scheduler using SQLAlchemy, patched for SQLAlchemy 2.x

Celery SQLAlchemy V2 Scheduler

A Celery Beat scheduler that stores the schedule in a database via SQLAlchemy.

This scheduler allows you to store periodic task schedules in a database, enabling you to add, edit, and remove tasks dynamically without restarting the Celery beat service.

---

Features

• Dynamic Schedule Management: Add, edit, and disable tasks on the fly by manipulating the database.
• SQLAlchemy Backend: Works with any database supported by SQLAlchemy (e.g., PostgreSQL, MySQL, SQLite).
• Full-Featured Schedules: Natively supports interval, crontab, and solar schedules.
• Timezone-Aware Crontabs: Define cron jobs that run in specific timezones.
• Declarative Setup: Define an initial schedule directly in your Celery configuration, which will be synchronized to the database on startup.

Installation

Install the package from PyPI:

pip install celery-sqlalchemy-v2-scheduler

Setup

To use this scheduler, you need to set the beat_scheduler and beat_dburi in your Celery application configuration. The scheduler will automatically create the necessary tables in your database when it first starts.

# in your_app/celery.py

from celery import Celery
from celery.schedules import crontab

app = Celery('your_app')

# Configure the scheduler
app.conf.beat_scheduler = 'celery_sqlalchemy_v2_scheduler.schedulers.DatabaseScheduler'

# The database URI for the scheduler.
# This can be any database supported by SQLAlchemy.
app.conf.beat_dburi = 'sqlite:///schedule.db'

# (Optional) Define a static schedule in your config.
# These tasks will be added to the database when the scheduler starts.
# This is useful for defining a default set of tasks.
app.conf.beat_schedule = {
    'cleanup-every-morning': {
        'task': 'your_app.tasks.backend_cleanup',
        'schedule': crontab(hour=4, minute=0),
    },
    'add-every-30-seconds': {
        'task': 'your_app.tasks.add',
        'schedule': 30.0,
        'args': (16, 16)
    },
}

# Load task modules
app.autodiscover_tasks(['your_app.tasks'])

Usage

Running Celery Beat

Start the beat service. If you have configured the scheduler in your Celery app as shown above, you don't need to specify it on the command line.

celery -A your_app beat -l info

The scheduler will connect to the database specified in beat_dburi and create the necessary tables if they don't exist.

Managing Tasks Programmatically

You can add, modify, or delete tasks by directly interacting with the SQLAlchemy models. This is the primary benefit of using a database-backed scheduler.

Here's an example of how to add a new periodic task that runs every 10 seconds.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from celery.schedules import schedule

from celery_sqlalchemy_v2_scheduler.session import SessionManager
from celery_sqlalchemy_v2_scheduler.models import PeriodicTask, IntervalSchedule

# 1. Setup the database session
db_uri = 'sqlite:///schedule.db'
session_manager = SessionManager()
engine, Session = session_manager.create_session(db_uri)
session = Session()

# 2. Create an interval schedule
# The scheduler will look for an existing schedule with the same properties
# or create a new one if it doesn't exist.
interval = IntervalSchedule.from_schedule(session, schedule(run_every=10.0))
session.flush() # Ensure the interval gets an ID

# 3. Create the periodic task
task = PeriodicTask(
    name='My Programmatic Task',
    task='your_app.tasks.some_task',
    interval=interval,
    args='[1, 2]',
    kwargs='{"foo": "bar"}',
    enabled=True
)

session.add(task)
session.commit()

print(f"Task '{task.name}' with id {task.id} created.")

session.close()

Disabling a Task

To disable a task, simply query it and set its enabled flag to False. The beat service will automatically detect the change and stop scheduling the task.

# ... (session setup from previous example) ...

task_to_disable = session.query(PeriodicTask).filter_by(name='My Programmatic Task').first()

if task_to_disable:
    task_to_disable.enabled = False
    session.commit()
    print(f"Task '{task_to_disable.name}' has been disabled.")

session.close()

Database Models

The scheduler uses the following core models:

• PeriodicTask: The main model representing a single periodic task. It holds the task name, arguments, execution options, and a foreign key to one of the schedule types.
• IntervalSchedule: Stores interval-based schedules (e.g., "run every 30 seconds").
• CrontabSchedule: Stores cron-style schedules (e.g., "run every day at 5 AM"). This model is timezone-aware.
• SolarSchedule: Stores schedules based on solar events like sunrise, sunset, dawn, and dusk for a given geographic location.
• PeriodicTaskChanged: A helper table used internally to efficiently detect when the schedule has been updated, prompting the scheduler to reload its tasks.