PMFlow/backend/app/services/operation_log_service.py

41 lines
990 B
Python

"""操作日志服务"""
from typing import Optional
from sqlalchemy.ext.asyncio import AsyncSession
from ..models import OperationLog
async def log_operation(
db: AsyncSession,
*,
tenant_id: int,
user_id: int,
project_id: Optional[int],
entity_type: str,
entity_id: str,
action: str,
before_data: Optional[dict] = None,
after_data: Optional[dict] = None,
remark: Optional[str] = None,
) -> None:
"""记录操作日志,失败时不影响主流程。"""
try:
db.add(
OperationLog(
tenant_id=tenant_id,
user_id=user_id,
project_id=project_id,
entity_type=entity_type,
entity_id=entity_id,
action=action,
before_data=before_data,
after_data=after_data,
remark=remark,
)
)
except Exception:
# 日志失败不阻断主交易
return