shanxia2.0/backend/app/core/oper_log_context.py

"""请求级操作日志元数据（ContextVar）"""
from __future__ import annotations

from contextvars import ContextVar
from dataclasses import dataclass
from typing import Optional


@dataclass
class OperLogMeta:
    business_title: Optional[str] = None
    business_type: Optional[int] = None
    business_id: Optional[str] = None
    oper_summary: Optional[str] = None


_oper_log_meta: ContextVar[Optional[OperLogMeta]] = ContextVar("oper_log_meta", default=None)


def set_oper_log_meta(meta: OperLogMeta) -> None:
    _oper_log_meta.set(meta)


def get_oper_log_meta() -> Optional[OperLogMeta]:
    return _oper_log_meta.get()


def clear_oper_log_meta() -> None:
    _oper_log_meta.set(None)


def snapshot_oper_log_meta() -> Optional[OperLogMeta]:
    """供异步落库线程使用，避免 ContextVar 跨线程丢失。"""
    return get_oper_log_meta()