我在Django中构建一个审计跟踪系统，需要在模型信号中访问当前的HTTP请求 (pre_save, post_save, pre_delete, post_delete)。

目标是让信号具备“请求感知能力”，以便我记录：

• 经过身份验证的用户 (request.user)
• IP地址
• 以及其他请求元数据

无需在每个服务层调用中显式传递请求对象。

由于Django信号本身不直接提供请求对象，我尝试使用 contextvars.ContextVar 来解决。下面是相关代码

# request_context.py
from contextvars import ContextVar

current_request = ContextVar("current_request", default=None)


def get_current_request():
    """Retrieve the current request from context."""
    return current_request.get()

# middleware.py
from core.request_context import current_request


class CurrentRequestMiddleware:
    """
    Stores the current HTTP request in a ContextVar
    so it can be accessed globally (e.g. inside signals).
    """

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

    def __call__(self, request):
        print("SETTING REQUEST:", request)

        token = current_request.set(request)

        try:
            response = self.get_response(request)
        finally:
            current_request.reset(token)

        return response

下面是在信号中执行它的位置

from core.request_context import get_current_request

request = get_current_request()

if request:
    print(request.__dict__)
else:
    print("Request is None")

预期行为

我期望的是因为：

• 中间件成功执行
• 请求通过ContextVar.set() 设置
• 信号在正常的请求生命周期中被触发（Django管理后台 / API请求）

那么信号中的get_current_request() 应返回活动的HttpRequest。但ContextVar.get() 在信号处理程序中返回None。现在我的问题是？

将Django信号“请求感知”以便在信号中可靠访问当前HttpRequest以实现审计日志的正确方式是什么？ 或者更好地， 在Django中实现请求感知审计日志的推荐生产安全模式是什么？

我知道使用Django信号进行审计日志记录通常被视为反模式，不建议用于复杂系统。 然而，在这个项目中，我受以下要求的限制：

• 我不能使用外部依赖或第三方审计库。
• 我需要一个内置的Django解决方案。
• 我必须将所有模型级操作（创建、更新、删除）记录到审计表。
• 系统应在可用时捕获请求上下文（用户、IP）等信息。审计日志必须自动完成，不需要在代码库中进行显式调用。

目标是确保对模型的所有Django ORM驱动操作都能集中追踪。

解决方案

我为你准备了一个可用的解决方案。我对你的代码进行了改进，并以我为FastAPI项目编写的中间件为模板进行建模。

# audit_context.py


from contextvars import ContextVar
from typing import Optional, Dict, Any

_audit_context: ContextVar[Optional[Dict[str, Any]]] = ContextVar("audit_context", default=None)


def set_audit_context(data: Dict[str, Any]) -> None:
    _audit_context.set(data)


def get_audit_context() -> Optional[Dict[str, Any]]:
    return _audit_context.get()

#  request_snapshot.py

import json

# This is to make sure not all META items are picked, given some of them are sensitive.
SAFE_META_PREFIXES = (
    "HTTP_",
    "REMOTE_ADDR",
    "CONTENT_TYPE",
    "CONTENT_LENGTH",
)


def build_request_snapshot(request):
    """
    Creates a FULL but SAFE serializable snapshot of Django request.
    """

    # --- HEADERS ---
    headers = dict(request.headers)

    # --- META (filtered) ---
    meta = {k: v for k, v in request.META.items() if k.startswith(SAFE_META_PREFIXES)}

    # --- BODY ---
    body = None
    try:
        raw = request.body

        if raw:
            text = raw.decode("utf-8", errors="ignore")

            # try JSON decode
            try:
                body = json.loads(text)
            except Exception:
                body = text

    except Exception as e:
        body = f"unavailable: {str(e)}"

    # --- COOKIES ---
    cookies = dict(request.COOKIES)

    # --- QUERY ---
    query_params = dict(request.GET)

    # --- USER ---
    user = request.user
    user_data = None
    if hasattr(user, "is_authenticated") and user.is_authenticated:
        user_data = {
            "id": getattr(user, "id", None),
            "username": getattr(user, "username", None),
            "email": getattr(user, "email", None),
        }

    return {
        "method": request.method,
        "path": request.path,
        "full_path": request.get_full_path(),
        "absolute_uri": request.build_absolute_uri(),
        "host": request.get_host(),
        "scheme": request.scheme,
        "ip": _get_client_ip(request),
        "headers": headers,
        "meta": meta,
        "query_params": query_params,
        "cookies": cookies,
        "body": body,
        "user": user_data,
    }


def _get_client_ip(request):
    xff = request.META.get("HTTP_X_FORWARDED_FOR")
    if xff:
        return xff.split(",")[0].strip()
    return request.META.get("REMOTE_ADDR")

# middlewares.py

from audit_context import set_audit_context
from request_snapshot import build_request_snapshot


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

    def __call__(self, request):

        snapshot = build_request_snapshot(request)

        set_audit_context(snapshot)

        return self.get_response(request)

不要忘记在settings.py中注册中间件

# my_signals.py

from django.db.models.signals import post_save
from django.dispatch import receiver

from models import OTP
from audit_context import get_audit_context


@receiver(post_save, sender=OTP)
def otp_post_save(sender, instance, created, **kwargs):

    ctx = get_audit_context()

    print("\n===== AUDIT LOG =====")
    print(ctx)
    print("=====================\n")

    if created:
        print("OTP CREATED")
    else:
        print("OTP UPDATED")

也不要忘记像往常一样注册你的信号。

将OTP模型替换为你正在使用的模型，我只是为了测试而使用了它。

你可以根据需要向build_request_snapshot函数的返回数据添加更多信息。

这做了什么：

• 请求进入；
• 它构建快照并保存它
• 请求-响应流继续进行
• 快照在你的信号中可用