tessl/pypi-django-celery-beat

Database-backed periodic task scheduling for Django and Celery integration

—

Pending

Overview

Eval results

Files

Validation and Utilities

Name: tessl/pypi-django-celery-beat
Author: tessl

Validation functions for cron expressions and utility functions for working with timezone-aware datetime objects and database-backed scheduling operations.

Capabilities

Cron Expression Validation

Comprehensive validation functions for cron expression fields, ensuring proper format and value ranges for minute, hour, day, month, and day-of-week fields.

def crontab_validator(value: str):
    """
    Validate a complete cron expression string.
    
    Parameters:
    - value (str): Complete cron expression to validate
    
    Raises:
    - ValidationError: If cron expression format is invalid
    """

def minute_validator(value: str):
    """
    Validate cron minute field (0-59).
    
    Parameters:
    - value (str): Minute field value to validate
    
    Raises:
    - ValidationError: If minute value is invalid
    """

def hour_validator(value: str):
    """
    Validate cron hour field (0-23).
    
    Parameters:
    - value (str): Hour field value to validate
    
    Raises:
    - ValidationError: If hour value is invalid
    """

def day_of_month_validator(value: str):
    """
    Validate cron day of month field (1-31).
    
    Parameters:
    - value (str): Day of month field value to validate
    
    Raises:
    - ValidationError: If day of month value is invalid
    """

def month_of_year_validator(value: str):
    """
    Validate cron month field (1-12).
    
    Parameters:
    - value (str): Month field value to validate
    
    Raises:
    - ValidationError: If month value is invalid
    """

def day_of_week_validator(value: str):
    """
    Validate cron day of week field (0-6).
    
    Parameters:
    - value (str): Day of week field value to validate
    
    Raises:
    - ValidationError: If day of week value is invalid
    """

Timezone Utilities

Utility functions for handling timezone-aware datetime objects in Django environments, ensuring proper timezone conversion and current time retrieval.

def make_aware(value: datetime.datetime) -> datetime.datetime:
    """
    Force datetime to have timezone information.
    
    Parameters:
    - value (datetime): Datetime object to make timezone-aware
    
    Returns:
    - datetime: Timezone-aware datetime object
    """

def now() -> datetime.datetime:
    """
    Return current date/time respecting Django timezone settings.
    
    Returns:
    - datetime: Current datetime with appropriate timezone
    """

def aware_now() -> datetime.datetime:
    """
    Return timezone-aware current datetime.
    
    Returns:
    - datetime: Current timezone-aware datetime
    """

Scheduler Utilities

Helper functions for working with database schedulers and checking scheduler configurations.

def is_database_scheduler(scheduler) -> bool:
    """
    Check if the provided scheduler is a database scheduler instance.
    
    Parameters:
    - scheduler: Scheduler instance to check
    
    Returns:
    - bool: True if scheduler is DatabaseScheduler instance
    """

Model Helper Functions

Utility functions used internally by the schedule models for cron expression formatting and timezone handling.

def cronexp(field) -> str:
    """
    Convert cron field to string representation.
    
    Parameters:
    - field: Cron field value to convert
    
    Returns:
    - str: String representation of cron field
    """

def crontab_schedule_celery_timezone() -> str:
    """
    Get timezone string from CELERY_TIMEZONE setting.
    
    Returns:
    - str: Timezone string (defaults to 'UTC')
    """

Constants

Timeout Constants

NEVER_CHECK_TIMEOUT = 100000000  # Large timeout value for disabled tasks

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"),
]

Usage Examples

Creating Validated Schedules

from django_celery_beat.models import CrontabSchedule
from django_celery_beat.validators import crontab_validator
from django.core.exceptions import ValidationError

# Validate before creating cron schedule
try:
    crontab_validator("0 2 * * 1")  # Every Monday at 2 AM
    schedule = CrontabSchedule.objects.create(
        minute='0',
        hour='2',
        day_of_week='1',
        day_of_month='*',
        month_of_year='*'
    )
except ValidationError as e:
    print(f"Invalid cron expression: {e}")

Working with Timezone-aware Datetimes

from django_celery_beat.utils import make_aware, now, aware_now
from datetime import datetime

# Get current time with proper timezone
current_time = now()
aware_time = aware_now()

# Make a naive datetime timezone-aware
naive_dt = datetime(2023, 12, 25, 15, 30)
aware_dt = make_aware(naive_dt)

Scheduler Detection

from django_celery_beat.utils import is_database_scheduler
from django_celery_beat.schedulers import DatabaseScheduler

scheduler = DatabaseScheduler()
if is_database_scheduler(scheduler):
    print("Using database scheduler")

Install with Tessl CLI

npx tessl i tessl/pypi-django-celery-beat

docs

admin-interface.md

database-scheduler.md

index.md

schedule-models.md

task-management.md

validation-utilities.md

tile.json