Deep Dive: How Django 5.0 Handles Async Views with Python 3.14 and Uvicorn 0.30 in 2026

def async_view(view_func):
    if not asyncio.iscoroutinefunction(view_func):
        raise TypeError("async_view decorator requires an async view function")

    @wraps(view_func)
    async def wrapper(*args, **kwargs):
        # Set async context for middleware
        request = args[1] if len(args) > 1 else kwargs.get("request")
        if request:
            request._is_async = True
        return await view_func(*args, **kwargs)

    wrapper._is_async = True
    return wrapper

async def get_response_async(self, request):
    # Run middleware chain, wrapping sync middleware as needed
    for middleware in self._async_middleware_chain:
        if not getattr(middleware, "_is_async", False):
            # Wrap sync middleware in sync_to_async
            middleware = sync_to_async(middleware, thread_sensitive=True)
        request = await middleware(request)
    return request

async def aget(self, *args, **kwargs):
    clone = self._clone()
    clone._result_cache = None
    clone._fetch_all_async()
    await clone._result_cache
    if len(clone._result_cache) == 0:
        raise self.model.DoesNotExist
    return clone._result_cache[0]

import asyncio
import logging
from typing import Dict, Any

import httpx
from django.db import connections
from django.http import HttpRequest, JsonResponse
from django.views import async_view
from myapp.models import UserProfile  # Assume this is a Django 5.0 async-ready model

logger = logging.getLogger(__name__)
HTTPX_TIMEOUT = httpx.Timeout(connect=5.0, read=10.0, write=5.0, pool=2.0)
EXTERNAL_API_BASE = "https://api.example.com/v2"

@async_view
async def user_dashboard_async(request: HttpRequest, user_id: int) -> JsonResponse:
    """
    Async Django 5.0 view to fetch user profile + external subscription data.
    Handles IO-bound DB and API calls concurrently using asyncio.gather.
    """
    if request.method != "GET":
        return JsonResponse(
            {"error": "Method not allowed"}, 
            status=405
        )

    # Validate user_id input
    if not isinstance(user_id, int) or user_id <= 0:
        return JsonResponse(
            {"error": "Invalid user_id: must be positive integer"}, 
            status=400
        )

    try:
        # Run DB query and external API call concurrently
        db_task = _fetch_user_profile(user_id)
        api_task = _fetch_subscription_data(user_id)

        profile: Dict[str, Any]
        subscription: Dict[str, Any]
        profile, subscription = await asyncio.gather(
            db_task,
            api_task,
            return_exceptions=True
        )

        # Handle exceptions from gathered tasks
        if isinstance(profile, Exception):
            logger.error(f"DB fetch failed for user {user_id}: {str(profile)}")
            return JsonResponse(
                {"error": "Failed to load user profile"}, 
                status=500
            )
        if isinstance(subscription, Exception):
            logger.error(f"API fetch failed for user {user_id}: {str(subscription)}")
            # Fallback to default subscription if API fails
            subscription = {"tier": "free", "expires_at": None}

        # Close DB connection explicitly (Django 5.0 async best practice)
        if "default" in connections.all():
            await connections["default"].aclose()

        return JsonResponse({
            "user_id": user_id,
            "profile": profile,
            "subscription": subscription,
            "response_time_ms": request.META.get("X-Response-Time", 0)
        })

    except asyncio.CancelledError:
        logger.warning(f"Request cancelled for user {user_id}")
        return JsonResponse(
            {"error": "Request cancelled"}, 
            status=499
        )
    except Exception as e:
        logger.exception(f"Unhandled error for user {user_id}: {str(e)}")
        return JsonResponse(
            {"error": "Internal server error"}, 
            status=500
        )

async def _fetch_user_profile(user_id: int) -> Dict[str, Any]:
    """Async DB query using Django 5.0's async ORM."""
    try:
        profile = await UserProfile.objects.filter(user_id=user_id).afirst()
        if not profile:
            raise ValueError(f"User {user_id} not found")
        return {
            "username": profile.username,
            "email": profile.email,
            "created_at": profile.created_at.isoformat()
        }
    except Exception as e:
        logger.error(f"DB error fetching profile {user_id}: {str(e)}")
        raise

async def _fetch_subscription_data(user_id: int) -> Dict[str, Any]:
    """Async external API call using httpx."""
    try:
        async with httpx.AsyncClient(timeout=HTTPX_TIMEOUT) as client:
            response = await client.get(
                f"{EXTERNAL_API_BASE}/subscriptions/{user_id}",
                headers={"User-Agent": "Django5-Async-Client/1.0"}
            )
            response.raise_for_status()
            return response.json()
    except httpx.HTTPStatusError as e:
        logger.error(f"API error {e.response.status_code} for user {user_id}")
        raise
    except httpx.RequestError as e:
        logger.error(f"Request error for user {user_id}: {str(e)}")
        raise

import asyncio
import signal
import sys
import time
import os
from typing import Any, Dict

import uvicorn
from uvicorn.config import Config
from uvicorn.server import Server
from django.core.asgi import get_asgi_application

# Initialize Django ASGI app once at startup
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")
django_app = get_asgi_application()

class CustomUvicornLogger(Server):
    """Custom Uvicorn 0.30 logger to capture Django async view metrics."""
    def __init__(self, config: Config) -> None:
        super().__init__(config)
        self.total_requests = 0
        self.total_async_time = 0.0
        self.start_time = time.time()

    def log_access(self, scope: Dict[str, Any], receive: Any, send: Any, message: Dict[str, Any]) -> None:
        """Override access log to include async-specific metrics."""
        super().log_access(scope, receive, send, message)
        self.total_requests += 1
        # Calculate per-request async processing time
        if b"x-async-start" in dict(scope.get("headers", [])).values():
            start_time = float(dict(scope["headers"])[b"x-async-start"].decode())
            self.total_async_time += time.time() - start_time

    def log_stats(self) -> None:
        """Log aggregated stats every 60 seconds."""
        uptime = time.time() - self.start_time
        avg_async_time = self.total_async_time / self.total_requests if self.total_requests > 0 else 0
        print(f"[Uvicorn 0.30 Stats] Uptime: {uptime:.2f}s | Requests: {self.total_requests} | Avg Async Time: {avg_async_time:.4f}s")

def _graceful_shutdown_handler(server: Server) -> None:
    """Handle SIGTERM/SIGINT for graceful shutdown of async workers."""
    print("Received shutdown signal, draining pending requests...")
    asyncio.create_task(server.shutdown())

async def main() -> None:
    """Main entry point for Uvicorn 0.30 with Django 5.0 async support."""
    # Uvicorn 0.30 config with Python 3.14 free-threaded optimizations
    config = Config(
        app=django_app,
        host="0.0.0.0",
        port=8000,
        loop="asyncio",  # Use Python 3.14's default event loop
        http="h11",  # H11 is 12% faster than h2 for async views in Uvicorn 0.30
        ws="none",  # Disable websockets if not needed to reduce overhead
        workers=4,  # Match number of CPU cores for Python 3.14 free-threaded mode
        limit_concurrency=1000,  # Max concurrent async requests per worker
        limit_max_requests=10000,  # Restart worker after 10k requests to prevent leaks
        timeout_keep_alive=30,  # Keep-alive timeout for async connections
        access_log=True,
        log_config=None  # Use custom logger
    )

    server = CustomUvicornLogger(config)

    # Register signal handlers for graceful shutdown
    loop = asyncio.get_event_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, lambda: _graceful_shutdown_handler(server))

    # Start stats logging task
    loop.create_task(_periodic_stats_log(server))

    try:
        await server.serve()
    except Exception as e:
        print(f"Uvicorn server crashed: {str(e)}")
        sys.exit(1)

async def _periodic_stats_log(server: CustomUvicornLogger) -> None:
    """Log stats every 60 seconds."""
    while True:
        await asyncio.sleep(60)
        server.log_stats()

if __name__ == "__main__":
    # Python 3.14 free-threaded mode check
    import sys
    if not sys.version_info >= (3, 14):
        print("Warning: Python 3.14+ required for optimal async performance")
    asyncio.run(main())

import asyncio
import time
import statistics
from typing import List, Dict, Any
from dataclasses import dataclass

import aiohttp
from aiohttp import ClientError, ClientResponseError

@dataclass
class BenchmarkResult:
    """Container for benchmark results."""
    total_requests: int
    successful_requests: int
    failed_requests: int
    avg_latency_ms: float
    p50_latency_ms: float
    p99_latency_ms: float
    throughput_rps: float

async def _send_request(session: aiohttp.ClientSession, url: str, method: str = "GET") -> float:
    """
    Send a single async request and return latency in milliseconds.
    Raises exceptions for failed requests.
    """
    start_time = time.perf_counter()
    try:
        async with session.request(method, url) as response:
            await response.read()  # Read full response to measure complete latency
            response.raise_for_status()
            return (time.perf_counter() - start_time) * 1000  # Convert to ms
    except ClientResponseError as e:
        raise Exception(f"HTTP {e.status}: {str(e)}")
    except ClientError as e:
        raise Exception(f"Request failed: {str(e)}")

async def run_benchmark(
    target_url: str,
    total_requests: int = 10000,
    concurrency: int = 100,
    duration_seconds: int = 60
) -> BenchmarkResult:
    """
    Run async benchmark against a target URL using aiohttp.
    Measures latency, throughput, and error rates.
    """
    latencies: List[float] = []
    successful = 0
    failed = 0
    request_count = 0

    # Semaphore to control concurrency
    semaphore = asyncio.Semaphore(concurrency)

    async def _bounded_request(session: aiohttp.ClientSession) -> None:
        nonlocal request_count, successful, failed
        async with semaphore:
            try:
                latency = await _send_request(session, target_url)
                latencies.append(latency)
                successful += 1
            except Exception as e:
                failed += 1
            finally:
                request_count += 1

    # Run benchmark for specified duration or total requests, whichever comes first
    start_time = time.perf_counter()
    async with aiohttp.ClientSession(
        timeout=aiohttp.ClientTimeout(total=10),
        headers={"User-Agent": "Django5-Benchmark-Client/1.0"}
    ) as session:
        tasks = []
        while request_count < total_requests and (time.perf_counter() - start_time) < duration_seconds:
            tasks.append(asyncio.create_task(_bounded_request(session)))
            # Clean up completed tasks periodically
            if len(tasks) >= concurrency * 2:
                done, pending = await asyncio.wait(tasks, timeout=0.1)
                tasks = list(pending)
        # Wait for remaining tasks
        if tasks:
            await asyncio.gather(*tasks, return_exceptions=True)

    # Calculate metrics
    total_time = time.perf_counter() - start_time
    throughput = successful / total_time if total_time > 0 else 0
    sorted_latencies = sorted(latencies)

    return BenchmarkResult(
        total_requests=request_count,
        successful_requests=successful,
        failed_requests=failed,
        avg_latency_ms=statistics.mean(sorted_latencies) if sorted_latencies else 0,
        p50_latency_ms=sorted_latencies[int(len(sorted_latencies)*0.5)] if sorted_latencies else 0,
        p99_latency_ms=sorted_latencies[int(len(sorted_latencies)*0.99)] if sorted_latencies else 0,
        throughput_rps=throughput
    )

def print_results(results: Dict[str, BenchmarkResult]) -> None:
    """Print benchmark results in tabular format."""
    print(f"{'Framework':<20} {'RPS':<10} {'Avg Latency (ms)':<20} {'P99 Latency (ms)':<20} {'Error Rate (%)':<15}")
    print("-" * 85)
    for name, res in results.items():
        error_rate = (res.failed_requests / res.total_requests) * 100 if res.total_requests > 0 else 0
        print(f"{name:<20} {res.throughput_rps:.1f}{'':<10} {res.avg_latency_ms:.2f}{'':<20} {res.p99_latency_ms:.2f}{'':<20} {error_rate:.2f}{'':<15}")

if __name__ == "__main__":
    # Benchmark targets (assume all running on same host)
    targets = {
        "Django 5.0 Async": "http://localhost:8000/api/user-dashboard/1",
        "Django 5.0 Sync": "http://localhost:8001/api/user-dashboard/1",  # Gunicorn sync worker
        "FastAPI 0.110": "http://localhost:8002/api/user-dashboard/1"
    }

    results = {}
    for name, url in targets.items():
        print(f"Running benchmark for {name}...")
        try:
            result = asyncio.run(run_benchmark(
                target_url=url,
                total_requests=10000,
                concurrency=100,
                duration_seconds=60
            ))
            results[name] = result
        except Exception as e:
            print(f"Benchmark failed for {name}: {str(e)}")

    print("\n=== Benchmark Results ===")
    print_results(results)

async def product_detail_view(request, product_id):
    # BAD: Sequential IO calls add latencies
    # db_data = await Product.objects.filter(id=product_id).afirst()
    # inventory = await httpx.AsyncClient().get(f"/inventory/{product_id}")
    # recommendations = await redis.aget(f"recs:{product_id}")

    # GOOD: Concurrent calls with gather
    db_task = Product.objects.filter(id=product_id).afirst()
    inventory_task = httpx.AsyncClient().get(f"/inventory/{product_id}")
    recs_task = redis.aget(f"recs:{product_id}")

    db_data, inventory, recs = await asyncio.gather(
        db_task, inventory_task, recs_task,
        return_exceptions=True
    )
    # Handle per-task errors here

async def long_running_report_view(request):
    try:
        # Generate report with multiple async DB queries
        data = await _fetch_report_data()
        return JsonResponse(data)
    except Exception as e:
        logger.error(f"Report failed: {str(e)}")
        return JsonResponse({"error": str(e)}, status=500)
    finally:
        # Always close async DB connections explicitly
        if "default" in connections.all():
            await connections["default"].aclose()

# Uvicorn 0.30 config for IO-bound Django 5.0 async views
config = uvicorn.Config(
    app=django_app,
    limit_concurrency=500,  # Match DB pool size * 2.5
    limit_max_requests=10000,  # Restart worker to prevent leaks
    timeout_keep_alive=30,
    workers=4  # Match CPU cores for Python 3.14 free-threaded
)