python
42 lines · 2 tabs
Priya Sharma
Jan 2026
2 tabs
from celery import Celery
from celery.schedules import crontab
app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.conf.beat_schedule = {
'send-daily-report': {
'task': 'reports.tasks.send_daily_report',
'schedule': crontab(hour=8, minute=0), # Every day at 8 AM
},
'cleanup-old-sessions': {
'task': 'core.tasks.cleanup_sessions',
'schedule': crontab(hour=2, minute=0, day_of_week=1), # Monday 2 AM
},
'refresh-cache': {
'task': 'cache.tasks.refresh_popular_items',
'schedule': 300.0, # Every 5 minutes
},
}
from celery import shared_task
from django.core.mail import send_mail
from django.utils import timezone
from datetime import timedelta
@shared_task
def send_daily_report():
"""Generate and send daily analytics report."""
yesterday = timezone.now() - timedelta(days=1)
# Generate report data
from analytics.models import DailyStats
stats = DailyStats.objects.filter(date=yesterday.date()).first()
if stats:
send_mail(
subject=f'Daily Report - {yesterday.date()}',
message=f'Users: {stats.user_count}, Revenue: ${stats.revenue}',
from_email='reports@example.com',
recipient_list=['admin@example.com'],
)
2 files · python
Explain with highlit
Celery Beat schedules periodic tasks like cron jobs. I define schedules in settings or use the Django database scheduler. Tasks run at specified intervals or cron expressions. I use @periodic_task decorator or configure in CELERY_BEAT_SCHEDULE. For dynamic schedules, I use django-celery-beat with database-backed periodic tasks. I monitor task execution and failures. Beat must run as a separate process alongside workers. This enables features like daily reports, cache warming, or cleanup jobs without external cron.
Related snips
python
import os
import stat
for root, _dirs, files in os.walk('/etc'):
for name in files:
path = os.path.join(root, name)Python security audit script for exposed risky filesystem state
python
auditing
host-security
by Kai Nakamura
1 tab
python
class Product(models.Model):
name = models.CharField(max_length=200)
slug = models.SlugField(blank=True)
price = models.DecimalField(max_digits=10, decimal_places=2)
cost = models.DecimalField(max_digits=10, decimal_places=2)
margin = models.DecimalField(max_digits=5, decimal_places=2, blank=True)Django model signals vs overriding save
django
python
models
by Priya Sharma
2 tabs
python
from django.urls import path
from . import views
app_name = 'blog'
urlpatterns = [Django URL namespacing and reverse lookups
django
python
urls
by Priya Sharma
3 tabs
python
import graphene
from graphene_django import DjangoObjectType
from blog.models import Post, Comment
class PostType(DjangoObjectType):Django GraphQL with Graphene
django
python
graphql
by Priya Sharma
2 tabs
python
from django.db.models import Count, Avg, Sum, Q, F
from django.views.generic import TemplateView
from products.models import Product, Order, OrderItem
class DashboardView(TemplateView):Django aggregation with annotate for statistics
django
python
database
by Priya Sharma
1 tab
python
from rest_framework import permissions
class IsOwner(permissions.BasePermission):
"""Allow only object owner to access."""Django REST Framework permissions and authorization
django
python
rest
by Priya Sharma
2 tabs
Share this code
Here's the card — post it anywhere.
