Python高级技术之：如何利用`SQLAlchemy`的`event`系统，在数据变更时触发自定义逻辑。

各位观众老爷，晚上好！我是你们的老朋友，今天咱们来聊聊Python里一个相当给力的工具：SQLAlchemy的event系统。这玩意儿就像一个隐藏的开关，允许你在数据库数据发生变化时，偷偷地塞入一些自定义逻辑，让你的代码更加灵活和强大。

Part 1: SQLAlchemy Event 系统是个啥？

简单来说，SQLAlchemy的event系统就像一个监听器，它可以监听数据库操作过程中的各种事件，比如对象被加载、被保存、被删除等等。一旦某个事件发生，你就可以通过预先注册的回调函数（也就是你自定义的逻辑）来执行一些额外的操作。

想象一下，你是一家电商网站的开发者。你希望在每次用户成功下单后，自动发送一封确认邮件。使用SQLAlchemy的event系统，你就可以在订单数据被成功插入数据库之后，触发一个事件，然后在这个事件的回调函数里，编写发送邮件的代码。是不是很酷？

Part 2: Event 系统的基本用法

要使用event系统，首先你需要导入sqlalchemy.event模块，然后使用listen()函数来注册事件监听器。listen()函数的基本语法如下：

from sqlalchemy import event

event.listen(target, identifier, fn, *args, **kwargs)

• target: 监听的目标对象。它可以是Engine、Connection、Session、Mapper或者一个具体的类。
• identifier: 要监听的事件的名称，比如before_insert、after_update、before_delete等等。
• fn: 回调函数。当事件发生时，这个函数会被调用。
• *args, **kwargs: 传递给回调函数的额外参数。

举个例子，假设我们有一个User类，我们想在每次User对象被插入数据库之前，自动设置它的创建时间：

from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import event
import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    created_at = Column(DateTime, default=datetime.datetime.utcnow)

    def __repr__(self):
        return f"<User(name='{self.name}', created_at='{self.created_at}')>"

# 定义事件监听函数
def before_insert_listener(mapper, connection, target):
    target.created_at = datetime.datetime.utcnow()  # 更新创建时间

# 创建一个简单的数据库连接
engine = create_engine('sqlite:///:memory:') # 使用内存数据库方便演示
Base.metadata.create_all(engine)

# 创建一个会话
Session = sessionmaker(bind=engine)
session = Session()

# 注册事件监听器
event.listen(User, 'before_insert', before_insert_listener)

# 创建一个 User 对象
new_user = User(name='Alice')

# 添加到会话并提交
session.add(new_user)
session.commit()

# 查询并打印用户信息
retrieved_user = session.query(User).filter_by(name='Alice').first()
print(retrieved_user)

session.close()

在这个例子中，我们定义了一个before_insert_listener函数，它会在User对象被插入数据库之前被调用。这个函数会将User对象的created_at属性设置为当前时间。

Part 3: 常用的 Event 类型

SQLAlchemy提供了各种各样的事件类型，你可以根据自己的需求选择合适的事件来监听。下面是一些常用的事件类型：

事件类型	描述	监听目标
before_insert	在对象被插入数据库之前触发。	Mapper, 类
after_insert	在对象被成功插入数据库之后触发。	Mapper, 类
before_update	在对象被更新到数据库之前触发。	Mapper, 类
after_update	在对象被成功更新到数据库之后触发。	Mapper, 类
before_delete	在对象被从数据库中删除之前触发。	Mapper, 类
after_delete	在对象被成功从数据库中删除之后触发。	Mapper, 类
before_flush	在Session的flush()方法被调用之前触发。	Session
after_flush	在Session的flush()方法被调用之后触发。	Session
before_commit	在Session的commit()方法被调用之前触发。	Session
after_commit	在Session的commit()方法被成功调用之后触发。	Session
before_rollback	在Session的rollback()方法被调用之前触发。	Session
after_rollback	在Session的rollback()方法被成功调用之后触发。	Session
connect	在Engine创建一个新的数据库连接时触发。	Engine
close	在数据库连接被关闭时触发。	Connection
before_execute	在SQL语句被执行之前触发。可以用来记录SQL语句或者修改SQL语句。这个事件在Connection级别触发。	Connection
after_execute	在SQL语句被执行之后触发。可以用来记录SQL语句的执行时间。这个事件在Connection级别触发。	Connection
load	在对象被从数据库中加载时触发。	Mapper, 类
refresh	在对象被刷新时（例如，使用session.refresh()方法）触发。	Mapper, 类

Part 4: 进阶用法：修改SQL语句，监听Engine事件

event系统不仅可以监听对象级别的事件，还可以监听Engine和Connection级别的事件。这使得你可以对SQL语句进行修改，或者记录SQL语句的执行时间。

4.1 修改SQL语句

有时候，你可能需要修改SQLAlchemy生成的SQL语句。例如，你可能需要在所有SELECT语句中添加一个WHERE子句，以实现多租户隔离。

from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    tenant_id = Column(Integer)

    def __repr__(self):
        return f"<User(name='{self.name}', tenant_id='{self.tenant_id}')>"

def add_tenant_id_to_query(statement, connection, **kw):
    # 只对select语句生效
    if not statement.is_select:
        return

    # 从会话中获取tenant_id (假设你把tenant_id存放在session的info属性中)
    tenant_id = connection.scalar(statement.compile(compile_kwargs={"literal_binds": True}).string) # 简单起见，这里我们直接获取tenant_id, 实际应用中要从session上下文中获取

    if tenant_id is not None:
        from sqlalchemy import text
        statement.where(User.tenant_id == tenant_id)

# 创建引擎
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

# 创建Session
Session = sessionmaker(bind=engine)

# 注册before_compile事件，修改SQL语句
event.listen(engine, "before_compile", add_tenant_id_to_query)

# 创建会话
session = Session()

# 创建一些数据
user1 = User(name='Alice', tenant_id=1)
user2 = User(name='Bob', tenant_id=2)
user3 = User(name='Charlie', tenant_id=1)
session.add_all([user1, user2, user3])
session.commit()

# 查询 tenant_id = 1 的用户
# session.info['tenant_id'] = 1 # 在session的info属性中存储 tenant_id，方便在事件监听函数中获取
# 模拟tenant_id = 1，直接硬编码
retrieved_users = session.query(User).filter(User.tenant_id == 1).all() # 显式添加tenant_id，对比一下结果
print("Users with tenant_id = 1:", retrieved_users)

# 查询 tenant_id = 2 的用户
# session.info['tenant_id'] = 2
# 模拟tenant_id = 2，直接硬编码
retrieved_users = session.query(User).filter(User.tenant_id == 2).all() # 显式添加tenant_id，对比一下结果
print("Users with tenant_id = 2:", retrieved_users)

session.close()

4.2 监听Engine事件

你可以使用connect事件来监听Engine创建新的数据库连接的事件。例如，你可以在每次创建新的数据库连接时，设置一些连接属性：

from sqlalchemy import create_engine, event

def set_sqlite_pragma(dbapi_connection, connection_record):
    cursor = dbapi_connection.cursor()
    cursor.execute("PRAGMA foreign_keys=ON")
    cursor.close()

engine = create_engine('sqlite:///:memory:')
event.listen(engine, 'connect', set_sqlite_pragma)

在这个例子中，我们使用connect事件来在每次创建新的SQLite数据库连接时，启用外键约束。

4.3 监听SQL执行事件

使用before_execute和after_execute事件，可以记录SQL语句的执行时间，方便性能分析。

import time
from sqlalchemy import create_engine, event

def before_execute(conn, clauseelement, multiparams, params):
    conn.info['query_start_time'] = time.time()
    print("Start Querying!")

def after_execute(conn, clauseelement, multiparams, params, result):
    total = time.time() - conn.info['query_start_time']
    print("Query took %f seconds" % total)

engine = create_engine('sqlite:///:memory:')
event.listen(engine, "before_execute", before_execute)
event.listen(engine, "after_execute", after_execute)

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))

Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

session.add(User(name='Alice'))
session.commit()

user = session.query(User).first()
print(user)

session.close()

这段代码会在每次SQL语句执行前后打印消息，并计算执行时间。

Part 5: Event 系统的注意事项

• 性能影响: event系统会增加一些额外的开销，因此不要过度使用。只在你真正需要的时候才使用它。
• 事务: event回调函数会在事务上下文中执行。如果在回调函数中发生错误，可能会导致事务回滚。
• 回调函数签名: event回调函数的签名必须与事件类型匹配。否则，SQLAlchemy会抛出一个异常。
• 递归: 要避免在回调函数中触发相同的事件，否则可能会导致无限递归。
• 上下文: 理解回调函数的上下文非常重要。 例如，在before_insert事件中，target参数是即将被插入数据库的对象。
• 调试: 调试event系统可能会比较困难。可以使用日志记录或者断点调试来帮助你找到问题。

Part 6: 实战案例：审计日志

一个常见的event系统应用场景是生成审计日志。你可以监听before_insert、before_update和before_delete事件，并将相关的操作信息记录到日志文件中。

from sqlalchemy import create_engine, Column, Integer, String, DateTime, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import event
import datetime
import logging

# 配置日志
logging.basicConfig(filename='audit.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    created_at = Column(DateTime, default=datetime.datetime.utcnow)

    def __repr__(self):
        return f"<User(name='{self.name}', created_at='{self.created_at}')>"

class AuditLog(Base):
    __tablename__ = 'audit_log'
    id = Column(Integer, primary_key=True)
    table_name = Column(String(50))
    record_id = Column(Integer)
    event_type = Column(String(20))
    old_values = Column(Text)
    new_values = Column(Text)
    created_at = Column(DateTime, default=datetime.datetime.utcnow)

# 定义审计日志记录函数
def log_changes(mapper, connection, target):
    table_name = target.__tablename__
    record_id = target.id

    # 记录插入事件
    if connection.scalar(f"SELECT COUNT(*) FROM {table_name} WHERE id = {record_id}") == 0:
         event_type = 'insert'
         new_values = {column.name: getattr(target, column.name) for column in target.__table__.columns}
         audit_log = AuditLog(table_name=table_name, record_id=record_id, event_type=event_type, new_values=str(new_values))
         connection.execute(AuditLog.__table__.insert().values(
            table_name=audit_log.table_name,
            record_id=audit_log.record_id,
            event_type=audit_log.event_type,
            new_values=audit_log.new_values,
            created_at=audit_log.created_at
        ))

    else:

        # 记录更新事件
        event_type = 'update'
        old_values = {}
        new_values = {}
        for column in target.__table__.columns:
            column_name = column.name
            old_value = connection.scalar(f"SELECT {column_name} FROM {table_name} WHERE id = {record_id}")
            new_value = getattr(target, column_name)
            if old_value != new_value:
                old_values[column_name] = old_value
                new_values[column_name] = new_value

        if old_values or new_values:
            audit_log = AuditLog(table_name=table_name, record_id=record_id, event_type=event_type, old_values=str(old_values), new_values=str(new_values))
            connection.execute(AuditLog.__table__.insert().values(
                table_name=audit_log.table_name,
                record_id=audit_log.record_id,
                event_type=audit_log.event_type,
                old_values=audit_log.old_values,
                new_values=audit_log.new_values,
                created_at=audit_log.created_at
            ))

# 注册事件监听器
event.listen(User, 'after_flush', log_changes) # 使用after_flush可以监听insert和update

# 创建一个简单的数据库连接
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

# 创建一个会话
Session = sessionmaker(bind=engine)
session = Session()

# 创建一个 User 对象
new_user = User(name='Alice')
session.add(new_user)
session.commit()

# 更新 User 对象
new_user.name = 'Bob'
session.commit()

# 删除 User 对象
session.delete(new_user)
session.commit()

# 打印审计日志
audit_logs = session.query(AuditLog).all()
for log in audit_logs:
    print(f"Table: {log.table_name}, Record ID: {log.record_id}, Event: {log.event_type}, Old Values: {log.old_values}, New Values: {log.new_values}")

session.close()

这个例子会在User对象被插入、更新或删除时，将相关的操作信息记录到audit_log表中。请注意，实际的审计日志记录可能需要更复杂的逻辑，例如记录操作用户的ID、IP地址等等。此外，日志记录最好异步处理，避免影响主流程的性能。

Part 7: 总结

SQLAlchemy的event系统是一个非常强大的工具，可以让你在数据库操作过程中执行自定义逻辑。你可以使用它来实现各种各样的功能，例如发送邮件、生成审计日志、缓存数据等等。但是，要记住合理使用event系统，避免过度使用，以免影响性能。

好了，今天的讲座就到这里。希望大家有所收获！下次再见！