如何在FastAPI中用MySQL实现审计日志（不依赖触发器）

1. 新增审计日志数据库模型 #

在你的models.py中添加审计日志表模型，覆盖所需的核心字段：

from sqlalchemy import Column, Integer, String, Text, DateTime, Enum
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
import enum

Base = declarative_base()

class OperationType(enum.Enum):
    CREATE = "CREATE"
    UPDATE = "UPDATE"
    DELETE = "DELETE"

class AuditLog(Base):
    __tablename__ = "audit_logs"
    
    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, index=True)  # 关联操作发起者的ID（可选）
    username = Column(String(50), nullable=False)
    operation_type = Column(Enum(OperationType), nullable=False)
    old_value = Column(Text)  # 存储旧数据的JSON字符串
    new_value = Column(Text)  # 存储新数据的JSON字符串
    operation_time = Column(DateTime, default=datetime.utcnow)

2. 编写审计日志工具函数 #

实现通用的日志保存函数，处理数据序列化与入库逻辑：

from sqlalchemy.orm import Session
from models import AuditLog, OperationType
import json
from datetime import datetime

def save_audit_log(
    db: Session,
    user_id: int,
    username: str,
    operation_type: OperationType,
    old_value: dict = None,
    new_value: dict = None
):
    # 将字典序列化为JSON字符串存储
    old_value_str = json.dumps(old_value) if old_value else None
    new_value_str = json.dumps(new_value) if new_value else None
    
    audit_log = AuditLog(
        user_id=user_id,
        username=username,
        operation_type=operation_type,
        old_value=old_value_str,
        new_value=new_value_str,
        operation_time=datetime.utcnow()
    )
    db.add(audit_log)
    db.commit()

3. 在增删改接口中调用审计日志 #

假设你已有get_current_user依赖用于获取当前登录用户，在各操作接口中嵌入日志逻辑：

创建用户接口示例 #

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from schemas import UserCreate, UserResponse
from models import User
from database import get_db
from utils import save_audit_log, OperationType
from auth import get_current_user

router = APIRouter(prefix="/users", tags=["users"])

@router.post("/", response_model=UserResponse)
def create_user(
    user_in: UserCreate,
    db: Session = Depends(get_db),
    current_user = Depends(get_current_user)
):
    db_user = User(**user_in.dict())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    
    # 记录创建操作日志（仅需新值）
    save_audit_log(
        db=db,
        user_id=current_user.id,
        username=current_user.username,
        operation_type=OperationType.CREATE,
        new_value=UserResponse.from_orm(db_user).dict()
    )
    
    return db_user

更新用户接口示例 #

@router.put("/{user_id}", response_model=UserResponse)
def update_user(
    user_id: int,
    user_in: UserCreate,
    db: Session = Depends(get_db),
    current_user = Depends(get_current_user)
):
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    
    # 保存旧数据（去除SQLAlchemy内部属性）
    old_value = db_user.__dict__.copy()
    del old_value["_sa_instance_state"]
    
    # 执行更新
    for key, value in user_in.dict().items():
        setattr(db_user, key, value)
    db.commit()
    db.refresh(db_user)
    
    # 保存新数据
    new_value = UserResponse.from_orm(db_user).dict()
    
    # 记录更新操作日志
    save_audit_log(
        db=db,
        user_id=current_user.id,
        username=current_user.username,
        operation_type=OperationType.UPDATE,
        old_value=old_value,
        new_value=new_value
    )
    
    return db_user

删除用户接口示例 #

@router.delete("/{user_id}")
def delete_user(
    user_id: int,
    db: Session = Depends(get_db),
    current_user = Depends(get_current_user)
):
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    
    # 保存待删除用户的数据
    old_value = db_user.__dict__.copy()
    del old_value["_sa_instance_state"]
    
    db.delete(db_user)
    db.commit()
    
    # 记录删除操作日志（仅需旧值）
    save_audit_log(
        db=db,
        user_id=current_user.id,
        username=current_user.username,
        operation_type=OperationType.DELETE,
        old_value=old_value
    )
    
    return {"detail": "User deleted successfully"}

4. 关键注意事项 #

• 当前用户获取：确保get_current_user能正确返回操作发起者的id和username，匿名操作场景可记录为"system"或留空。
• 数据序列化：优先使用Pydantic模型转换数据（如UserResponse.from_orm(db_user).dict()），避免SQLAlchemy内部属性被序列化。
• 性能优化：高并发场景可改用异步任务（如Celery）异步写入日志，普通场景同步写入即可满足需求。
• 日志查询：可额外实现审计日志查询接口，支持按用户名、操作类型、时间范围过滤。

内容的提问来源于stack exchange，提问作者AMITA DHANDHA