35 lines
868 B
Python
35 lines
868 B
Python
"""请求级操作日志元数据（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()
|