celery-mongobeat 0.1.4

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.

Why celery-mongobeat?

The original celerybeat-mongo library is no longer actively maintained and contains several critical bugs. This project was created to provide a stable, reliable, and modern alternative for the community, ensuring continued support for dynamic, database-backed Celery schedules.

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: Supports legacy configuration variables from celerybeat-mongo for a smoother transition.
• 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.beat:MongoScheduler"

Migrating from celerybeat-mongo

celery-mongobeat is designed as a near drop-in replacement, but there is one important configuration change you must make when migrating:

• Update the Scheduler Path: The import path for the scheduler has been updated to align with modern package structures and Celery best practices.

You must change your beat_scheduler setting from: 'celerybeat_mongo.schedulers.MongoScheduler' (the old path) to: 'celery_mongobeat.beat:MongoScheduler' (the new path)

### Legacy (Backwards-Compatible) Configuration

If you are migrating from `celerybeat-mongo`, this library provides backward compatibility for the uppercase configuration variables. Modern, lowercase settings (e.g., `mongodb_scheduler_url`) will always take precedence.

```python
# celeryconfig.py

# Legacy uppercase individual settings (from celerybeat-mongo)
CELERY_MONGODB_SCHEDULER_URL = "mongodb://localhost:27017/"
CELERY_MONGODB_SCHEDULER_DB = "celery"
CELERY_MONGODB_SCHEDULER_COLLECTION = "schedules"

beat_scheduler = "celery_mongobeat.beat: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
)