本文目录导读：
在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)
推荐使用方案
对于大多数项目，我推荐使用方案1（事件监听器），它具有以下优点：
- 完全透明：不需要修改业务代码
- 灵活控制：可以只对特定表或特定字段启用
- 易于定制：可以添加自定义字段（如用户IP、请求ID等）
- 性能可控：可以批量写入或异步处理
选择建议：
- 简单项目：使用方案1（事件监听器）
- 需要更强大的功能：使用方案4（sqlalchemy-continuum）
- 对性能要求极高：使用方案3（数据库触发器）
- 需要特定审计逻辑：使用方案2（混合类）
根据你的具体需求选择合适的方案即可。
