tessl/pypi-django-celery-beat
Database-backed periodic task scheduling for Django and Celery integration
—
Pending
schedule-models.mddocs/
Schedule Models
Database models that define when periodic tasks should execute, supporting multiple scheduling patterns including intervals, cron expressions, solar events, and one-time execution.
Capabilities
Interval Scheduling
Schedule tasks to run at regular intervals with configurable period types.
class IntervalSchedule(models.Model):
"""
Schedule that runs at a specific interval.
Fields:
- every: int - Number of periods between runs (minimum 1)
- period: str - Type of period (DAYS, HOURS, MINUTES, SECONDS, MICROSECONDS)
"""
every: int
period: str
# Period constants
DAYS = 'days'
HOURS = 'hours'
MINUTES = 'minutes'
SECONDS = 'seconds'
MICROSECONDS = 'microseconds'
PERIOD_CHOICES = [
(DAYS, 'Days'),
(HOURS, 'Hours'),
(MINUTES, 'Minutes'),
(SECONDS, 'Seconds'),
(MICROSECONDS, 'Microseconds'),
]
@property
def schedule(self) -> schedules.schedule: ...
@property
def period_singular(self) -> str: ...
@classmethod
def from_schedule(cls, schedule: schedules.schedule) -> 'IntervalSchedule': ...Usage Example:
from django_celery_beat.models import IntervalSchedule
# Create a schedule that runs every 30 minutes
schedule = IntervalSchedule.objects.create(
every=30,
period=IntervalSchedule.MINUTES
)
# Or get existing schedule
schedule, created = IntervalSchedule.objects.get_or_create(
every=5,
period=IntervalSchedule.SECONDS
)Cron-style Scheduling
Timezone-aware cron-like scheduling with support for complex timing patterns.
class CrontabSchedule(models.Model):
"""
Cron-like schedule with timezone support.
Fields follow cron format: minute hour day_of_month month_of_year day_of_week
Each field accepts cron expressions (*, numbers, ranges, lists).
"""
minute: str # 0-59, max 240 chars, with validators.minute_validator
hour: str # 0-23, max 96 chars, with validators.hour_validator
day_of_week: str # 0-6 (Sunday=0), max 64 chars, with validators.day_of_week_validator
day_of_month: str # 1-31, max 124 chars, with validators.day_of_month_validator
month_of_year: str # 1-12, max 64 chars, with validators.month_of_year_validator
timezone: timezone_field.TimeZoneField
@property
def schedule(self) -> Union[TzAwareCrontab, crontab]: ...
@property
def human_readable(self) -> str: ...
@classmethod
def from_schedule(cls, schedule: crontab) -> 'CrontabSchedule': ...
def due_start_time(self, initial_start_time: datetime.datetime, tz: Optional[tzinfo.tzinfo] = None) -> datetime.datetime: ...Usage Example:
from django_celery_beat.models import CrontabSchedule
import zoneinfo
# Run at 2:30 AM every day in Pacific timezone
schedule = CrontabSchedule.objects.create(
minute='30',
hour='2',
day_of_week='*',
day_of_month='*',
month_of_year='*',
timezone=zoneinfo.ZoneInfo('America/Los_Angeles')
)
# Run every 15 minutes
schedule = CrontabSchedule.objects.create(
minute='*/15',
hour='*',
day_of_week='*',
day_of_month='*',
month_of_year='*',
timezone=zoneinfo.ZoneInfo('UTC')
)Solar Event Scheduling
Schedule tasks based on astronomical events like sunrise, sunset, and solar noon for specific geographic locations.
class SolarSchedule(models.Model):
"""
Schedule following astronomical patterns.
Fields:
- event: str - Solar event type (sunrise, sunset, solar_noon, etc.)
- latitude: decimal.Decimal - Geographic latitude (-90 to 90)
- longitude: decimal.Decimal - Geographic longitude (-180 to 180)
"""
event: str
latitude: decimal.Decimal
longitude: decimal.Decimal
# Solar event choices
SOLAR_SCHEDULES = [
('dawn_astronomical', 'Astronomical dawn'),
('dawn_civil', 'Civil dawn'),
('dawn_nautical', 'Nautical dawn'),
('dusk_astronomical', 'Astronomical dusk'),
('dusk_civil', 'Civil dusk'),
('dusk_nautical', 'Nautical dusk'),
('solar_noon', 'Solar noon'),
('sunrise', 'Sunrise'),
('sunset', 'Sunset'),
]
@property
def schedule(self) -> schedules.solar: ...
@classmethod
def from_schedule(cls, schedule: schedules.solar) -> 'SolarSchedule': ...Usage Example:
from django_celery_beat.models import SolarSchedule
from decimal import Decimal
# Run at sunrise in New York City
schedule = SolarSchedule.objects.create(
event='sunrise',
latitude=Decimal('40.7128'),
longitude=Decimal('-74.0060') # Note: negative for west longitude
)
# Run at sunset in London
schedule = SolarSchedule.objects.create(
event='sunset',
latitude=Decimal('51.5074'),
longitude=Decimal('-0.1278')
)One-time Scheduling
Schedule tasks to run exactly once at a specific date and time.
class ClockedSchedule(models.Model):
"""
Schedule for one-time execution at a specific datetime.
Fields:
- clocked_time: datetime.datetime - When the task should run
"""
clocked_time: datetime.datetime
@property
def schedule(self) -> clocked: ...
@classmethod
def from_schedule(cls, schedule: clocked) -> 'ClockedSchedule': ...Usage Example:
from django_celery_beat.models import ClockedSchedule
from datetime import datetime, timedelta
# Run once in 1 hour
future_time = datetime.utcnow() + timedelta(hours=1)
schedule = ClockedSchedule.objects.create(
clocked_time=future_time
)
# Run at a specific time
specific_time = datetime(2024, 12, 25, 9, 0, 0) # Christmas 9 AM
schedule = ClockedSchedule.objects.create(
clocked_time=specific_time
)Model Helper Functions
Cron Expression Formatting
def cronexp(field: Optional[str]) -> str:
"""
Convert cron field to string representation.
Parameters:
- field: Cron field value or None
Returns:
str: String representation, '*' if field is None/empty
"""Timezone Detection
def crontab_schedule_celery_timezone() -> str:
"""
Get timezone string from Django CELERY_TIMEZONE setting.
Returns:
str: Timezone string, defaults to 'UTC' if not set or invalid
"""Constants
Period Constants
DAYS = 'days'
HOURS = 'hours'
MINUTES = 'minutes'
SECONDS = 'seconds'
MICROSECONDS = 'microseconds'
PERIOD_CHOICES = [
(DAYS, 'Days'),
(HOURS, 'Hours'),
(MINUTES, 'Minutes'),
(SECONDS, 'Seconds'),
(MICROSECONDS, 'Microseconds'),
]
SINGULAR_PERIODS = [
(DAYS, 'Day'),
(HOURS, 'Hour'),
(MINUTES, 'Minute'),
(SECONDS, 'Second'),
(MICROSECONDS, 'Microsecond'),
]Solar Event Constants
SOLAR_SCHEDULES = [
('dawn_astronomical', 'Astronomical dawn'),
('dawn_civil', 'Civil dawn'),
('dawn_nautical', 'Nautical dawn'),
('dusk_astronomical', 'Astronomical dusk'),
('dusk_civil', 'Civil dusk'),
('dusk_nautical', 'Nautical dusk'),
('solar_noon', 'Solar noon'),
('sunrise', 'Sunrise'),
('sunset', 'Sunset'),
]Types
Custom Schedule Types
class clocked(BaseSchedule):
"""
Schedule for one-time execution at specific datetime.
"""
def __init__(self, clocked_time: datetime.datetime): ...
def is_due(self, last_run_at: datetime.datetime) -> tuple[bool, float]: ...
def remaining_estimate(self, last_run_at: datetime.datetime) -> datetime.timedelta: ...
class TzAwareCrontab(crontab):
"""
Timezone-aware cron scheduling.
"""
def __init__(self, minute='*', hour='*', day_of_week='*',
day_of_month='*', month_of_year='*',
tz: Optional[tzinfo.tzinfo] = None): ...
def is_due(self, last_run_at: datetime.datetime) -> tuple[bool, float]: ...
def nowfunc(self) -> datetime.datetime: ...Install with Tessl CLI
npx tessl i tessl/pypi-django-celery-beat