Как реализовать асинхронность в Django?

Асинхронность в Django

Джанго исторически был синхронным фреймворком, но начиная с версии 3.1 поддерживает асинхронный код. Это позволяет обрабатывать больше одновременных подключений без использования многопроцессности.

1. Асинхронные views (Django 3.1+)

Базовый пример

# views.py
import asyncio
from django.http import JsonResponse

# Просто добавь async перед def
async def get_user_async(request, user_id):
    # Это автоматически запустится в event loop
    user = await fetch_user_from_db(user_id)
    return JsonResponse({"id": user.id, "name": user.name})

# Маршрут
from django.urls import path

urlpatterns = [
    path('users/<int:user_id>/', get_user_async),  # async view работает!
]

Что внутри происходит

Когда Django видит async view:

1. Запрос приходит
2. Django создаёт event loop (если его нет)
3. Запускает async view в этом loop
4. Ждёт завершения
5. Отправляет ответ

2. Асинхронные запросы в БД

Problem: Django ORM не async по умолчанию

# ❌ НЕПРАВИЛЬНО: ORM операции блокируют
async def get_user(user_id):
    user = User.objects.get(id=user_id)  # БЛОКИРУЕТ!
    return user

# Это синхронная операция, она заморозит весь event loop!

Решение 1: sync_to_async

from django.utils.decorators import sync_to_async
from django.db.models import F

# Оборни синхронный код в async
@sync_to_async
def get_user_from_db(user_id):
    return User.objects.get(id=user_id)

async def user_view(request, user_id):
    user = await get_user_from_db(user_id)  # Теперь async!
    return JsonResponse({"name": user.name})

Решение 2: Асинхронный ORM (asyncpg, databases)

Для более сложных случаев используй асинхронный драйвер:

pip install asyncpg databases

import databases
from typing import List

# Инициализация
database = databases.Database(
    "postgresql+asyncpg://user:pass@localhost/db"
)

# Асинхронная функция для запросов
async def get_users() -> List[dict]:
    query = "SELECT id, name, email FROM users"
    return await database.fetch_all(query)

# View
async def users_list(request):
    users = await get_users()
    return JsonResponse({"users": users})

3. Асинхронные функции-помощники

Пример: HTTP запрос

import aiohttp
import asyncio
from django.http import JsonResponse

async def fetch_external_api(url: str):
    """Асинхронный запрос к внешнему API"""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

async def dashboard_view(request):
    # Параллельно запрашиваем несколько API
    github_data, api_data = await asyncio.gather(
        fetch_external_api("https://api.github.com/users/python"),
        fetch_external_api("https://api.example.com/data")
    )
    
    return JsonResponse({
        "github": github_data,
        "api": api_data
    })

Асинхронные tasks

import asyncio
from django.http import JsonResponse

async def send_emails_async(user_ids: list):
    """Отправляем письма параллельно"""
    tasks = []
    for user_id in user_ids:
        tasks.append(send_email_to_user(user_id))
    
    results = await asyncio.gather(*tasks)
    return results

async def notify_users(request):
    user_ids = [1, 2, 3, 4, 5]
    await send_emails_async(user_ids)  # Параллельно!
    return JsonResponse({"status": "emails sent"})

4. Асинхронные Middleware

# middleware.py
import time
from django.utils.decorators import sync_and_async_middleware

@sync_and_async_middleware
def timing_middleware(get_response):
    if asyncio.iscoroutinefunction(get_response):
        async def middleware(request):
            start = time.time()
            response = await get_response(request)
            response["X-Process-Time"] = time.time() - start
            return response
    else:
        def middleware(request):
            start = time.time()
            response = get_response(request)
            response["X-Process-Time"] = time.time() - start
            return response
    
    return middleware

5. Асинхронные сигналы и обработчики

from django.dispatch import receiver
from django.core.signals import request_started
from django.utils.decorators import sync_and_async_middleware
import asyncio

# Асинхронный обработчик сигнала
@receiver(request_started, dispatch_uid='async_signal')
async def handle_request_started(sender, **kwargs):
    # Выполняем асинхронный код
    await log_request_async(kwargs)

6. Использование Celery для async tasks

Для тяжёлых операций лучше использовать task queue:

pip install celery redis

# celery.py
from celery import Celery
import os
import django

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
django.setup()

app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

# tasks.py
from celery import shared_task
import time

@shared_task
def send_email_task(user_id):
    """Задача запущена асинхронно в Celery worker"""
    user = User.objects.get(id=user_id)
    send_email(user.email)
    return f"Email sent to {user.email}"

# views.py
from django.http import JsonResponse
from .tasks import send_email_task

def send_notification(request, user_id):
    # Запускаем задачу в фоне
    task = send_email_task.delay(user_id)
    return JsonResponse({"task_id": task.id})

def get_task_status(request, task_id):
    from celery.result import AsyncResult
    task = AsyncResult(task_id)
    return JsonResponse({"status": task.status, "result": task.result})

7. Асинхронный WebSocket (Django Channels)

Для real-time функционала используй Django Channels:

pip install channels channels-redis

# settings.py
INSTALLED_APPS = [
    'daphne',  # ASGI сервер вместо WSGI
    'channels',
    # остальные приложения
]

ASGI_APPLICATION = 'myproject.asgi.application'

# asgi.py
import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from django.urls import path
from . import consumers

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

application = ProtocolTypeRouter({
    'http': get_asgi_application(),
    'websocket': AuthMiddlewareStack(
        URLRouter([
            path('ws/chat/<room_name>/', consumers.ChatConsumer.as_asgi()),
        ])
    ),
})

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

class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.room_name = self.scope['url_route']['kwargs']['room_name']
        self.room_group_name = f'chat_{self.room_name}'
        
        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name
        )
        
        await self.accept()
    
    async def receive(self, text_data):
        data = json.loads(text_data)
        message = data['message']
        
        # Отправляем всем в группе
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                'type': 'chat_message',
                'message': message
            }
        )
    
    async def chat_message(self, event):
        message = event['message']
        
        await self.send(text_data=json.dumps({
            'message': message
        }))

8. Асинхронная миграция с ASGI

# Используй Daphne или Hypercorn вместо Gunicorn
pip install daphne

# Запуск
daphne -b 0.0.0.0 -p 8000 myproject.asgi:application

# Или с несколькими воркерами
daphne -b 0.0.0.0 -p 8000 -w 4 myproject.asgi:application

9. Конфигурация для production

# settings.py

# Для ASGI сервера (async views)
ASGI_APPLICATION = 'myproject.asgi.application'

# Если используешь Celery для async tasks
CELERY_BROKER_URL = 'redis://localhost:6379'
CELERY_RESULT_BACKEND = 'redis://localhost:6379'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'

# Database оптимизация
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'ATOMIC_REQUESTS': False,  # Отключи для async
        'CONN_MAX_AGE': 0,  # Не переиспользуй соединения
    }
}

# Асинхронные фичи
DATABASES['default']['OPTIONS'] = {
    'connect_timeout': 10,
}

10. Best Practices

✅ Делай так:

# 1. Используй async для I/O операций
async def fetch_data(url):
    async with aiohttp.ClientSession() as session:
        return await session.get(url)

# 2. Используй sync_to_async для ORM
get_user = sync_to_async(User.objects.get)

# 3. Параллели операции с asyncio.gather
results = await asyncio.gather(task1(), task2(), task3())

# 4. Используй Celery для long-running tasks
@shared_task
def long_operation():
    ...

❌ Не делай так:

# 1. Не блокируй event loop
async def view(request):
    time.sleep(5)  # ПЛОХО!
    return response

# 2. Не мешай sync и async без adapter
async def view(request):
    user = User.objects.get(id=1)  # Заморозит!

# 3. Не запускай async в синхронном коде
# без asyncio.run()

Заключение

Асинхронность в Django — это evolution, не revolution. Варианты:

• async views — для простых I/O операций
• sync_to_async — обёртка для ORM
• Celery — для тяжелых фоновых задач
• Channels — для WebSocket и real-time
• ASGI — инфраструктура для всего выше

Выбирай инструмент в зависимости от задачи!