SQLAlchemy审计日志怎么自动记录

wen python案例 2026-07-12 181

本文目录导读：

使用事件监听器（推荐方案）
使用混合类（Mixin）方案
使用数据库触发器方案
高级方案：使用SQLAlchemy-Continuum
推荐使用方案

在SQLAlchemy中实现自动审计日志（记录数据变更历史）有几种主要方法，这里提供几种常用方案的实现代码：

使用事件监听器（推荐方案）

这是最灵活、最常用的方法，通过监听 SQLAlchemy 的事件来实现。

from sqlalchemy import create_engine, Column, Integer, String, DateTime, event
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
import json
Base = declarative_base()
# 审计日志表
class AuditLog(Base):
    __tablename__ = 'audit_logs'
    id = Column(Integer, primary_key=True)
    table_name = Column(String(100))
    record_id = Column(Integer)
    action = Column(String(20))  # INSERT, UPDATE, DELETE
    old_values = Column(String(1000))  # JSON格式存储
    new_values = Column(String(1000))  # JSON格式存储
    changed_by = Column(String(100))
    changed_at = Column(DateTime, default=datetime.utcnow)
# 示例业务表
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String(50))
    email = Column(String(100))
    age = Column(Integer)
# 创建审计日志的装饰器
def audit_log(target_class, exclude_columns=None):
    if exclude_columns is None:
        exclude_columns = ['id', 'created_at', 'updated_at']
    @event.listens_for(target_class, 'after_insert')
    def after_insert(mapper, connection, target):
        create_audit_log(
            connection,
            target_class.__tablename__,
            target.id,
            'INSERT',
            {},
            get_changed_values(target, exclude_columns),
            get_current_user()
        )
    @event.listens_for(target_class, 'after_update')
    def after_update(mapper, connection, target):
        # 获取旧值（需要从session中获取）
        session = Session.object_session(target)
        if session:
            old_values = {}
            for attr in target.__table__.columns.keys():
                if attr not in exclude_columns:
                    # 获取变更前的值
                    history = session.get_history(target, attr)
                    if history.has_changes():
                        old_values[attr] = history.deleted[0] if history.deleted else None
            if old_values:  # 只在有变化时记录
                create_audit_log(
                    connection,
                    target_class.__tablename__,
                    target.id,
                    'UPDATE',
                    old_values,
                    get_changed_values(target, exclude_columns),
                    get_current_user()
                )
    @event.listens_for(target_class, 'before_delete')
    def before_delete(mapper, connection, target):
        old_values = get_all_values(target, exclude_columns)
        create_audit_log(
            connection,
            target_class.__tablename__,
            target.id,
            'DELETE',
            old_values,
            {},
            get_current_user()
        )
def create_audit_log(connection, table_name, record_id, action, old_values, new_values, changed_by):
    audit_entry = AuditLog(
        table_name=table_name,
        record_id=record_id,
        action=action,
        old_values=json.dumps(old_values, default=str),
        new_values=json.dumps(new_values, default=str),
        changed_by=changed_by,
        changed_at=datetime.utcnow()
    )
    connection.execute(audit_entry.__table__.insert().values(
        table_name=audit_entry.table_name,
        record_id=audit_entry.record_id,
        action=audit_entry.action,
        old_values=audit_entry.old_values,
        new_values=audit_entry.new_values,
        changed_by=audit_entry.changed_by,
        changed_at=audit_entry.changed_at
    ))
def get_changed_values(target, exclude_columns):
    """获取模型实例的所有非排除字段值"""
    return {
        col.name: getattr(target, col.name)
        for col in target.__table__.columns
        if col.name not in exclude_columns
    }
def get_all_values(target, exclude_columns):
    """获取模型实例的所有值"""
    return {
        col.name: getattr(target, col.name)
        for col in target.__table__.columns
        if col.name not in exclude_columns
    }
def get_current_user():
    """获取当前用户（需要根据你的认证机制实现）"""
    # 示例：从Flask的request中获取
    # from flask import request
    # return getattr(request, 'user', 'anonymous')
    return 'anonymous'
# 应用审计日志到指定模型
audit_log(User, exclude_columns=['id'])
# 数据库初始化
engine = create_engine('sqlite:///example.db')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

使用混合类（Mixin）方案

from sqlalchemy import Column, Integer, DateTime, event
from sqlalchemy.orm import declarative_mixin
from datetime import datetime
import json
class AuditLogMixin:
    """审计日志混合类"""
    def create_audit_entry(self, action, session):
        from sqlalchemy import inspect
        from sqlalchemy.orm import class_mapper
        # 获取表名和主键
        mapper = class_mapper(self.__class__)
        table_name = mapper.tables[0].name
        primary_key = inspect(self).identity[0]
        # 获取变更前后的值
        old_values = {}
        new_values = {}
        for attr in mapper.attrs:
            if isinstance(attr, Column):
                col_name = attr.key
                if col_name not in ['id', 'created_at', 'updated_at']:
                    history = inspect(self).attrs[col_name].history
                    if history.has_changes():
                        old_values[col_name] = history.deleted[0] if history.deleted else None
                        new_values[col_name] = history.added[0] if history.added else None
        # 创建审计日志
        audit_log = {
            'table_name': table_name,
            'record_id': primary_key,
            'action': action,
            'old_values': json.dumps(old_values, default=str),
            'new_values': json.dumps(new_values, default=str),
            'changed_by': 'system',
            'changed_at': datetime.utcnow()
        }
        return audit_log
# 使用方法
class UserWithAudit(Base, AuditLogMixin):
    __tablename__ = 'users_with_audit'
    id = Column(Integer, primary_key=True)
    username = Column(String(50))
    email = Column(String(100))
    age = Column(Integer)

使用数据库触发器方案

-- PostgreSQL触发器示例
CREATE TABLE audit_logs (
    id SERIAL PRIMARY KEY,
    table_name VARCHAR(100),
    record_id INTEGER,
    action VARCHAR(20),
    old_values JSONB,
    new_values JSONB,
    changed_by VARCHAR(100),
    changed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE OR REPLACE FUNCTION audit_trigger_function()
RETURNS TRIGGER AS $$
BEGIN
    IF (TG_OP = 'INSERT') THEN
        INSERT INTO audit_logs(table_name, record_id, action, old_values, new_values, changed_by)
        VALUES (TG_TABLE_NAME, NEW.id, 'INSERT', '{}'::jsonb, row_to_json(NEW), current_user);
        RETURN NEW;
    ELSIF (TG_OP = 'UPDATE') THEN
        INSERT INTO audit_logs(table_name, record_id, action, old_values, new_values, changed_by)
        VALUES (TG_TABLE_NAME, NEW.id, 'UPDATE', row_to_json(OLD), row_to_json(NEW), current_user);
        RETURN NEW;
    ELSIF (TG_OP = 'DELETE') THEN
        INSERT INTO audit_logs(table_name, record_id, action, old_values, new_values, changed_by)
        VALUES (TG_TABLE_NAME, OLD.id, 'DELETE', row_to_json(OLD), '{}'::jsonb, current_user);
        RETURN OLD;
    END IF;
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;
-- 为需要审计的表添加触发器
CREATE TRIGGER users_audit_trigger
AFTER INSERT OR UPDATE OR DELETE ON users
FOR EACH ROW EXECUTE FUNCTION audit_trigger_function();

高级方案：使用SQLAlchemy-Continuum

这是一个专门的审计日志库,提供了更多功能：

# 安装: pip install sqlalchemy-continuum
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy_continuum import make_versioned
from sqlalchemy_continuum.plugins import PropertyModTrackerPlugin
# 启用版本控制
make_versioned(plugins=[PropertyModTrackerPlugin()])
class User(Base):
    __tablename__ = 'users'
    __versioned__ = {}
    id = Column(Integer, primary_key=True)
    username = Column(String(50))
    email = Column(String(100))
    age = Column(Integer)
# 创建数据库
engine = create_engine('sqlite:///example.db')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)