or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

admin-interface.mddatabase-scheduler.mdindex.mdschedule-models.mdtask-management.mdvalidation-utilities.md
tile.json

tessl/pypi-django-celery-beat

Database-backed periodic task scheduling for Django and Celery integration

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/django-celery-beat@2.8.x

To install, run

npx @tessl/cli install tessl/pypi-django-celery-beat@2.8.0

index.mddocs/

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(): ...