如何用SQLAlchemy为FastAPI项目实现数据库审计日志？

我希望在Python的FastAPI项目中创建审计日志功能。我已通过SQLAlchemy创建了若干数据表，希望为这些表构建审计日志以记录操作历史。当数据表执行插入、更新或删除操作时，需将操作日志捕获至指定审计表中。请问如何通过SQLAlchemy实现该需求？

原代码存在的问题 #

• 拼写错误：connection.exicute 应为 connection.execute
• 事件名称格式错误："after update" 需改为 "after_update"（下划线分隔）
• 监听对象不统一：混用Usermodel和user，需保持模型类名一致
• 函数名重复：两个日志处理函数同名update_user_log，会导致覆盖
• 缺少核心审计字段（如操作时间、操作人等，可按需补充）

修正后的实现方案 #

1. 先定义审计日志表结构

确保你已在models.py中定义业务表和审计日志表：

# models.py
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

# 业务表示例：用户表
class UserModel(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), unique=True, nullable=False)
    # 其他业务字段...

# 审计日志表：用户操作日志
class UserLog(Base):
    __tablename__ = "user_logs"
    id = Column(Integer, primary_key=True, autoincrement=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    action = Column(String(10), nullable=False)  # 记录操作类型：insert/update/delete
    created_at = Column(DateTime, default=datetime.utcnow)
    # 可选扩展字段：operator_id（操作人ID）、operator_ip（操作IP）等

2. 编写SQLAlchemy事件监听逻辑

通过SQLAlchemy的事件监听机制，在业务表操作后自动写入审计日志：

from sqlalchemy import event
from models import UserModel, UserLog

# 监听用户插入操作
@event.listens_for(UserModel, "after_insert")
def log_user_insert(mapper, connection, target):
    insert_stmt = UserLog.__table__.insert().values(
        user_id=target.id,
        action="insert"
    )
    connection.execute(insert_stmt)

# 监听用户更新操作
@event.listens_for(UserModel, "after_update")
def log_user_update(mapper, connection, target):
    insert_stmt = UserLog.__table__.insert().values(
        user_id=target.id,
        action="update"
    )
    connection.execute(insert_stmt)

# 监听用户删除操作
@event.listens_for(UserModel, "after_delete")
def log_user_delete(mapper, connection, target):
    insert_stmt = UserLog.__table__.insert().values(
        user_id=target.id,
        action="delete"
    )
    connection.execute(insert_stmt)

3. 扩展到多表的通用方案

如果需要给多个业务表添加审计日志，可以封装通用函数减少重复代码：

from datetime import datetime

def create_audit_log(connection, log_table, target_id, action):
    """通用审计日志插入函数"""
    insert_stmt = log_table.__table__.insert().values(
        target_id=target_id,
        action=action,
        created_at=datetime.utcnow()
    )
    connection.execute(insert_stmt)

# 为用户表绑定事件
@event.listens_for(UserModel, "after_insert")
def log_user_insert(mapper, connection, target):
    create_audit_log(connection, UserLog, target.id, "insert")

# 为其他业务表（如订单表）绑定事件示例
# from models import OrderModel, OrderLog
# @event.listens_for(OrderModel, "after_update")
# def log_order_update(mapper, connection, target):
#     create_audit_log(connection, OrderLog, target.id, "update")

核心说明 #

• SQLAlchemy的after_insert/after_update/after_delete事件会在对应操作成功完成后触发，确保日志记录与业务操作一致性
• 使用connection.execute直接执行SQL语句，避免ORM层额外开销
• 若需记录操作人、操作IP等信息，可结合FastAPI的请求上下文（from fastapi import Request），通过全局变量或依赖注入传递上下文数据（注意线程安全）

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