Как работать с асинхронными запросами в Django?python-52

pip install django channels daphne

import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter

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

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    # Дополнительные протоколы (WebSocket) можно добавить здесь
})

INSTALLED_APPS = [
    ...
    'channels',
]

ASGI_APPLICATION = 'myproject.asgi.application'

from django.http import JsonResponse
from asgiref.sync import sync_to_async
import httpx

async def async_view(request):
    # Асинхронный HTTP-запрос
    async with httpx.AsyncClient() as client:
        response = await client.get('https://api.example.com/data')
        data = response.json()

    return JsonResponse(data)

from django.contrib.auth.models import User
from asgiref.sync import sync_to_async

@sync_to_async
def get_users_count():
    return User.objects.count()

async def user_count_view(request):
    count = await get_users_count()
    return JsonResponse({'count': count})

from django.http import JsonResponse
from myapp.models import Product

async def product_list(request):
    products = await Product.objects.all().alist()
    return JsonResponse({'products': list(products)})

from django.db import transaction

async def create_product(request):
    async with transaction.atomic():
        product = await Product.objects.acreate(
            name="New Product",
            price=100.00
        )
    return JsonResponse({'id': product.id})

import asyncio
import httpx

async def fetch_data(url):
    async with httpx.AsyncClient() as client:
        return await client.get(url)

async def multiple_requests_view(request):
    urls = [
        'https://api.example.com/data1',
        'https://api.example.com/data2'
    ]

    tasks = [fetch_data(url) for url in urls]
    responses = await asyncio.gather(*tasks)

    data = [resp.json() for resp in responses]
    return JsonResponse({'results': data})

async def async_middleware(get_response):
    async def middleware(request):
        # Предобработка запроса
        print("Async before view")

        response = await get_response(request)

        # Постобработка ответа
        print("Async after view")
        return response

    return middleware

from django.test import TestCase
from httpx import AsyncClient

class AsyncTestCase(TestCase):
    async def test_async_view(self):
        client = AsyncClient()
        response = await client.get('/async-url/')
        self.assertEqual(response.status_code, 200)