Skip to content

Advanced Django#

Signals#

# signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import User, Profile

@receiver(post_save, sender=User)
def create_profile(sender, instance, created, **kwargs):
    if created:
        Profile.objects.create(user=instance)

Middleware#

# middleware.py
class SimpleMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        # Code before view
        response = self.get_response(request)
        # Code after view
        return response

Celery Task#

# tasks.py
from celery import shared_task

@shared_task
def send_email_task(email):
    # Heavy task here
    pass

Django Channels (WebSocket)#

# consumers.py
import json
from channels.generic.websocket import AsyncWebsocketConsumer

class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        await self.accept()

    async def receive(self, text_data):
        await self.send(text_data=json.dumps({'message': 'pong'}))

Database Transactions#

from django.db import transaction

@transaction.atomic
def complex_view(request):
    # All or nothing database operations
    pass

Caching#

from django.core.cache import cache

# Set cache
cache.set('key', 'value', timeout=300)

# Get cache
value = cache.get('key')