CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-django-celery-beat

Database-backed periodic task scheduling for Django and Celery integration

—

Pending
Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

SecuritybySnyk

Pending

The risk profile of this skill

Overview
Eval results
Files

Django Celery Beat

Django Celery Beat provides database-backed periodic task scheduling for Django applications using Celery. It enables storing and managing periodic tasks in a Django database, allowing dynamic task scheduling through the Django Admin interface without requiring server restarts or configuration file changes.

Package Information

  • Package Name: django-celery-beat
  • Language: Python
  • Installation: pip install django-celery-beat

Core Imports

from django_celery_beat.models import PeriodicTask, IntervalSchedule, CrontabSchedule
from django_celery_beat.schedulers import DatabaseScheduler

Basic Usage

from django_celery_beat.models import PeriodicTask, IntervalSchedule
import json

# Create an interval schedule (every 10 seconds)
schedule, created = IntervalSchedule.objects.get_or_create(
    every=10,
    period=IntervalSchedule.SECONDS,
)

# Create a periodic task
PeriodicTask.objects.create(
    interval=schedule,
    name='Import contacts every 10 seconds',
    task='myapp.tasks.import_contacts',
    args=json.dumps(['arg1', 'arg2']),
    kwargs=json.dumps({'be_careful': True}),
)

Architecture

Django Celery Beat integrates with Django's ORM and Celery's beat scheduler:

  • Django Models: Store schedule definitions and task configurations in database tables
  • Database Scheduler: Replaces Celery's default file-based scheduler with database queries
  • Admin Integration: Provides Django Admin interface for managing tasks
  • Signal System: Automatically tracks changes to notify the beat service of schedule updates

Capabilities

Schedule Models

Database models for defining when periodic tasks should run, including interval-based, cron-like, solar event-based, and one-time scheduling patterns.

class IntervalSchedule(models.Model):
    every: int
    period: str  # DAYS, HOURS, MINUTES, SECONDS, MICROSECONDS
    
class CrontabSchedule(models.Model):
    minute: str
    hour: str
    day_of_week: str
    day_of_month: str
    month_of_year: str
    timezone: timezone_field.TimeZoneField

class SolarSchedule(models.Model):
    event: str  # sunrise, sunset, solar_noon, etc.
    latitude: decimal.Decimal
    longitude: decimal.Decimal

class ClockedSchedule(models.Model):
    clocked_time: datetime.datetime

Schedule Models

Task Management

Core model and utilities for defining and managing periodic tasks, including task configuration, arguments, routing, and execution tracking.

class PeriodicTask(models.Model):
    name: str
    task: str
    interval: ForeignKey[IntervalSchedule]
    crontab: ForeignKey[CrontabSchedule]
    solar: ForeignKey[SolarSchedule]
    clocked: ForeignKey[ClockedSchedule]
    args: str  # JSON encoded list
    kwargs: str  # JSON encoded dict
    enabled: bool
    one_off: bool

class PeriodicTasks(models.Model):
    @classmethod
    def changed(cls): ...
    @classmethod
    def update_changed(cls): ...

Task Management

Database Scheduler

Celery beat scheduler implementation that reads periodic tasks from the database instead of configuration files, enabling dynamic task management.

class DatabaseScheduler(Scheduler):
    def setup_schedule(self): ...
    def all_as_schedule(self) -> dict: ...
    def schedule_changed(self) -> bool: ...
    def sync(self): ...

class ModelEntry(ScheduleEntry):
    def is_due(self) -> tuple[bool, float]: ...
    def save(self): ...

Database Scheduler

Admin Interface

Django Admin integration providing web-based management interface for periodic tasks and schedules with custom forms, actions, and validation.

class PeriodicTaskAdmin(ModelAdmin):
    def enable_tasks(self, request, queryset): ...
    def disable_tasks(self, request, queryset): ...
    def toggle_tasks(self, request, queryset): ...
    def run_tasks(self, request, queryset): ...

class TaskSelectWidget(Select):
    def tasks_as_choices(self) -> list[tuple[str, str]]: ...

Admin Interface

Validation and Utilities

Validation functions for cron expressions, timezone utilities, and helper functions for working with database-backed scheduling.

def crontab_validator(value: str): ...
def minute_validator(value: str): ...
def hour_validator(value: str): ...
def day_of_month_validator(value: str): ...
def month_of_year_validator(value: str): ...
def day_of_week_validator(value: str): ...

def make_aware(value: datetime.datetime) -> datetime.datetime: ...
def now() -> datetime.datetime: ...
def aware_now() -> datetime.datetime: ...
def is_database_scheduler(scheduler) -> bool: ...

Validation and Utilities

Django App Configuration

Django application configuration for integrating django-celery-beat with Django projects.

class BeatConfig(AppConfig):
    name: str
    label: str  
    verbose_name: str
    default_auto_field: str
    
    def ready(self): ...

Custom QuerySet

Enhanced QuerySet for PeriodicTask with optimization methods.

class PeriodicTaskQuerySet(QuerySet):
    def enabled(self) -> 'PeriodicTaskQuerySet': ...

Signal Management

Signal handling for automatic change tracking when tasks are modified.

def signals_connect(): ...
Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/django-celery-beat@2.8.x
Publish Source
CLI
Badge
tessl/pypi-django-celery-beat badge