Django Middleware Nâng Cao 2026: Custom Middleware, Logging và Câu Hỏi Phỏng Vấn

# settings.py
MIDDLEWARE = [
    "django.middleware.security.SecurityMiddleware",
    "django.contrib.sessions.middleware.SessionMiddleware",
    "django.middleware.common.CommonMiddleware",
    "django.middleware.csrf.CsrfViewMiddleware",
    "django.contrib.auth.middleware.AuthenticationMiddleware",
    "django.contrib.messages.middleware.MessageMiddleware",
    "django.middleware.clickjacking.XFrameOptionsMiddleware",
]

# middleware/timing.py
import time
import logging

logger = logging.getLogger(__name__)

class RequestTimingMiddleware:
    """Ghi lại thời gian xử lý mỗi request."""

    def __init__(self, get_response):
        # Được gọi một lần khi server khởi động
        self.get_response = get_response

    def __call__(self, request):
        # Code trước khi view thực thi
        start_time = time.perf_counter()

        # Gọi middleware tiếp theo hoặc view
        response = self.get_response(request)

        # Code sau khi view thực thi
        duration_ms = (time.perf_counter() - start_time) * 1000
        logger.info(
            "Request %s %s hoàn thành trong %.2fms",
            request.method,
            request.path,
            duration_ms
        )

        return response

# middleware/async_timing.py
import time
import logging
from asgiref.sync import iscoroutinefunction, markcoroutinefunction

logger = logging.getLogger(__name__)

class AsyncRequestTimingMiddleware:
    """Async-native timing middleware cho deployment ASGI."""

    async_capable = True
    sync_capable = False

    def __init__(self, get_response):
        self.get_response = get_response
        if iscoroutinefunction(self.get_response):
            markcoroutinefunction(self)

    async def __call__(self, request):
        start_time = time.perf_counter()

        # Await middleware tiếp theo hoặc async view
        response = await self.get_response(request)

        duration_ms = (time.perf_counter() - start_time) * 1000
        logger.info(
            "Async request %s %s hoàn thành trong %.2fms",
            request.method,
            request.path,
            duration_ms
        )

        return response

# middleware/request_id.py
import uuid
import logging

class RequestIDMiddleware:
    """Gắn request ID duy nhất vào mỗi request để tương quan log."""

    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        # Tạo hoặc trích xuất request ID từ header
        request_id = request.headers.get("X-Request-ID")
        if not request_id:
            request_id = str(uuid.uuid4())

        # Gắn vào object request để truy cập trong views
        request.request_id = request_id

        response = self.get_response(request)

        # Thêm request ID vào response headers
        response["X-Request-ID"] = request_id

        return response

# logging_filters.py
import logging

class RequestIDFilter(logging.Filter):
    """Chèn request_id vào log records."""

    def filter(self, record):
        from django.middleware.request_id import local
        record.request_id = getattr(local, "request_id", "no-request-id")
        return True

# settings.py
LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "filters": {
        "request_id": {
            "()": "logging_filters.RequestIDFilter",
        },
    },
    "formatters": {
        "verbose": {
            "format": "[{levelname}] {asctime} [{request_id}] {name}: {message}",
            "style": "{",
        },
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "formatter": "verbose",
            "filters": ["request_id"],
        },
    },
    "root": {
        "handlers": ["console"],
        "level": "INFO",
    },
}

# middleware/permission_check.py
from django.http import HttpResponseForbidden

class PermissionCheckMiddleware:
    """Kiểm tra quyền trước khi thực thi view."""

    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        return self.get_response(request)

    def process_view(self, request, view_func, view_args, view_kwargs):
        # Truy cập metadata của view
        required_permission = getattr(view_func, "required_permission", None)

        if required_permission:
            if not request.user.has_perm(required_permission):
                return HttpResponseForbidden("Permission denied")

        # Return None để tiếp tục đến view
        return None

# views.py
def require_permission(perm):
    def decorator(view_func):
        view_func.required_permission = perm
        return view_func
    return decorator

@require_permission("app.can_edit")
def edit_resource(request, pk):
    # Logic view
    pass

# middleware/error_tracking.py
import logging
import traceback

logger = logging.getLogger(__name__)

class ErrorTrackingMiddleware:
    """Ghi lại exception với đầy đủ ngữ cảnh trước khi Django xử lý."""

    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        return self.get_response(request)

    def process_exception(self, request, exception):
        logger.error(
            "Unhandled exception trong %s %s: %s\n%s",
            request.method,
            request.path,
            str(exception),
            traceback.format_exc(),
            extra={
                "user_id": getattr(request.user, "id", None),
                "request_id": getattr(request, "request_id", None),
            }
        )
        # Return None để Django xử lý exception
        return None

# settings.py
MIDDLEWARE = [
    # ... middleware khác
    "django.contrib.auth.middleware.LoginRequiredMiddleware",
]

# Miễn trừ view cụ thể
from django.contrib.auth.decorators import login_not_required

@login_not_required
def public_view(request):
    return HttpResponse("Nội dung công khai")

# middleware/legacy_style.py
from django.middleware import MiddlewareMixin

class LegacyStyleMiddleware(MiddlewareMixin):
    """Middleware sử dụng mẫu hook-based legacy."""

    def process_request(self, request):
        # Được gọi trước view, return None để tiếp tục
        request.custom_attribute = "value"
        return None

    def process_response(self, request, response):
        # Được gọi sau view, phải return response
        response["X-Custom-Header"] = "middleware-added"
        return response