celery-mongobeat 0.1.3

A Celery Beat scheduler that stores the schedule in MongoDB.

Celery-MongoBeat

A modern, drop-in replacement for celerybeat-mongo. This project provides a Celery Beat scheduler that stores and retrieves task schedules from a MongoDB collection, allowing for dynamic management of periodic tasks without restarting the Celery Beat service.

Features

• Stable and Reliable: Fixes critical bugs from celerybeat-mongo, such as the issue where disabling one task would prevent all tasks from running.
• Dynamic Task Management: Add, modify, and remove periodic tasks on the fly without restarting the beat service.
• MongoDB Backend: Leverages MongoDB for a robust and scalable schedule store.
• Fine-Grained Control:
  • Run Count Limiting: Use max_run_count to run a task a specific number of times and then automatically disable it.
• Flexible Configuration: Full support for advanced pymongo.MongoClient options (like SSL) via mongodb_scheduler_client_kwargs.
• Backwards Compatible: Designed as a drop-in replacement for the deprecated celerybeat-mongo. It supports the legacy mongodb_backend_settings configuration.
• Modern Tooling: Built with a modern Python packaging structure (pyproject.toml).
• All Schedule Types: Natively supports interval, crontab, and solar schedules.

Installation

Install the package from PyPI:

pip install celery-mongobeat

Configuration

To use this scheduler, set the beat_scheduler option in your Celery configuration.

Recommended Configuration

# celeryconfig.py

mongodb_scheduler_url = "mongodb://localhost:27017/"
mongodb_scheduler_db = "celery"
mongodb_scheduler_collection = "schedules"

beat_scheduler = "celery_mongobeat.schedulers:MongoScheduler"

Legacy (Backwards-Compatible) Configuration

If you are migrating from celerybeat-mongo, you can use your existing configuration.

# celeryconfig.py

mongodb_backend_settings = {
    "host": "mongodb://localhost:27017/",
    "database": "celery",
    "collection": "schedules"
}

beat_scheduler = "celery_mongobeat.schedulers:MongoScheduler"

Usage

Once configured, start Celery Beat as you normally would:

celery -A your_app beat -l info

You can now manage your schedules by adding, updating, or removing documents in the configured MongoDB collection.

Programmatic Usage Example

While you can manage schedules by inserting raw documents into MongoDB, it's often cleaner to use a helper class within your application. Here is an example of a ScheduleManager class that you could use in your project to programmatically create, update, and find tasks.

This example is framework-agnostic and uses pymongo directly.

from pymongo.collection import Collection

class ScheduleManager:
    """A helper class to manage schedule entries in MongoDB."""

    def __init__(self, collection: Collection):
        self.collection = collection

    def create_interval_task(self, name: str, task: str, every: int, period: str = 'seconds', args=None, kwargs=None, max_run_count: int = None):
        """Creates a task that runs on a fixed interval."""
        args = args or []
        kwargs = kwargs or {}
        schedule_doc = {
            'name': name,
            'task': task,
            'enabled': True,
            'interval': {'every': every, 'period': period},
            'args': args,
            'kwargs': kwargs,
        }
        if max_run_count is not None:
            schedule_doc['max_run_count'] = max_run_count

        self.collection.update_one(
            {'name': name},
            {'$set': schedule_doc},
            upsert=True
        )
        print(f"Upserted interval task: '{name}'")

    def create_crontab_task(self, name: str, task: str, minute='*', hour='*', day_of_week='*', args=None, kwargs=None):
        """Creates a task that runs on a crontab schedule."""
        args = args or []
        kwargs = kwargs or {}
        schedule_doc = {
            'name': name,
            'task': task,
            'enabled': True,
            'crontab': {'minute': minute, 'hour': hour, 'day_of_week': day_of_week},
            'args': args,
            'kwargs': kwargs,
        }
        self.collection.update_one(
            {'name': name},
            {'$set': schedule_doc},
            upsert=True
        )
        print(f"Upserted crontab task: '{name}'")

    def disable_task(self, name: str):
        """Disables a task by its unique name."""
        self.collection.update_one(
            {'name': name},
            {'$set': {'enabled': False}}
        )
        print(f"Disabled task: '{name}'")

You could then use this manager in your application like so:

# In your application's setup code
from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017/")
db = client["celery"]
schedules_collection = db["schedules"] # Must match your celery-mongobeat config

manager = ScheduleManager(schedules_collection)

# Example: Create a task to run every 30 seconds
manager.create_interval_task(
    name='my-periodic-task',
    task='your_project.tasks.some_task',
    every=30,
    period='seconds',
    args=[1, 2, 3]
)

# Example: Create a task that runs 5 times and then stops
manager.create_interval_task(
    name='run-five-times-task',
    task='your_project.tasks.some_task',
    every=60,
    period='seconds',
    max_run_count=5
)