SQLAlchemy ORM实战：Python数据库操作最佳实践-开发者社区

1. 为什么选择SQLAlchemy ORM

在Python生态中操作数据库，我们通常面临几个选择：直接使用数据库驱动（如psycopg2、mysql-connector）、使用DB-API这样的底层接口，或者采用ORM框架。SQLAlchemy作为Python中最成熟的ORM工具之一，它完美平衡了灵活性和易用性。

我最初接触SQLAlchemy是在一个需要快速迭代的Web项目里。当时团队尝试过直接写SQL，但随着业务复杂度增加，维护成百上千条SQL语句变成了噩梦。字段变更导致的连锁反应、不同数据库方言的兼容问题、手写JOIN查询时遗漏关联条件...这些问题在采用SQLAlchemy后得到了显著改善。

SQLAlchemy的核心优势在于它的分层设计：

最底层提供Engine处理数据库连接和方言转换
中间层的SQL Expression Language允许你像写Python代码一样构建SQL
顶层的ORM将数据库表映射为Python类

这种设计让你可以在需要精细控制时直接操作SQL，而在常规业务场景中享受ORM的便利。比如当我们需要优化一个复杂报表查询时，可以绕过ORM直接使用SQLAlchemy Core，而在处理用户注册、订单创建等标准CRUD操作时，使用ORM能节省大量重复代码。

2. 环境准备与安装指南

2.1 安装SQLAlchemy核心包

安装SQLAlchemy只需要一条简单的pip命令：

pip install sqlalchemy

但根据我的经验，在实际项目中还需要考虑以下因素：

版本锁定：生产环境应该固定版本以避免意外升级导致兼容问题。推荐使用pip install sqlalchemy==1.4.36这样的明确版本号。

数据库驱动：SQLAlchemy需要对应的数据库驱动：

# PostgreSQL pip install psycopg2-binary # 开发环境推荐 pip install psycopg2 # 生产环境推荐 # MySQL pip install mysql-connector-python # 官方驱动 pip install pymysql # 纯Python实现 # SQLite (Python内置支持)

注意：生产环境避免使用psycopg2-binary，它包含预编译的二进制文件可能引发兼容性问题。应该通过系统包管理器安装依赖后编译安装psycopg2。

2.2 开发环境配置建议

我习惯在项目中创建一个专门的db.py文件来集中管理数据库配置：

# db.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker DATABASE_URL = "postgresql://user:password@localhost:5432/mydb" # echo=True会在控制台输出SQL，开发环境非常有用 engine = create_engine( DATABASE_URL, echo=True, pool_size=5, # 连接池大小 max_overflow=10, # 允许超出pool_size的临时连接数 pool_timeout=30, # 获取连接超时时间(秒) pool_recycle=3600 # 连接回收时间(秒) ) SessionLocal = sessionmaker( autocommit=False, autoflush=False, bind=engine )

这样设计有几个好处：

数据库配置集中管理，修改时只需调整一个文件
连接池参数可调优，避免连接泄漏
开发时通过echo=True可以直观看到生成的SQL

3. 数据模型定义的艺术

3.1 基础模型定义

定义模型是使用ORM的第一步，也是最重要的一步。SQLAlchemy 2.x推荐使用declarative_base方式：

from sqlalchemy import Column, Integer, String from sqlalchemy.orm import declarative_base Base = declarative_base() class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True) username = Column(String(50), unique=True, nullable=False) email = Column(String(100), unique=True, index=True) hashed_password = Column(String(100))

这里有几个经验点：

__tablename__最好明确指定，避免依赖类名自动转换
字符串字段要设置合理长度，避免浪费存储空间
唯一约束(unique)和索引(index)能显著提升查询性能
密码等敏感字段应该存储哈希值而非明文

3.2 关系建模实战

现实中的数据很少是孤立的，SQLAlchemy提供了强大的关系表达能力。假设我们要构建一个博客系统：

from sqlalchemy import ForeignKey from sqlalchemy.orm import relationship class Post(Base): __tablename__ = 'posts' id = Column(Integer, primary_key=True) title = Column(String(100), nullable=False) content = Column(String(5000)) author_id = Column(Integer, ForeignKey('users.id')) # 定义多对一关系 author = relationship("User", back_populates="posts") # 定义多对多关系 tags = relationship("Tag", secondary="post_tags", back_populates="posts") class Tag(Base): __tablename__ = 'tags' id = Column(Integer, primary_key=True) name = Column(String(30), unique=True) posts = relationship("Post", secondary="post_tags", back_populates="tags") # 关联表(多对多关系的中间表) class PostTag(Base): __tablename__ = 'post_tags' post_id = Column(Integer, ForeignKey('posts.id'), primary_key=True) tag_id = Column(Integer, ForeignKey('tags.id'), primary_key=True)

关键点说明：

back_populates参数建立了双向关系，使得可以从User访问其Posts，也可以从Post访问其Author
多对多关系需要额外的关联表，这里我们显式定义了PostTag类以便添加额外字段(如创建时间)
关系可以是延迟加载(默认)或立即加载，大型系统要注意N+1查询问题

4. 会话管理与CRUD操作

4.1 会话生命周期管理

SQLAlchemy的Session是数据库交互的核心接口。我推荐使用上下文管理器模式管理会话：

from contextlib import contextmanager from sqlalchemy.orm import Session @contextmanager def get_db(): db = SessionLocal() try: yield db db.commit() except Exception: db.rollback() raise finally: db.close() # 使用示例 with get_db() as db: new_user = User(username="johndoe", email="john@example.com") db.add(new_user) # 这里不需要显式commit，上下文管理器会自动处理

这种模式有几个优势：

确保会话总是被正确关闭，避免连接泄漏
自动处理事务提交和回滚
代码更简洁，不需要到处写try-catch块

4.2 增删改查最佳实践

创建数据：

# 单个创建 with get_db() as db: user = User(username="alice", email="alice@example.com") db.add(user) # 此时user.id为None，因为还未插入数据库 db.flush() # 执行INSERT但不提交事务 print(user.id) # 现在有id了 # 批量创建(更高效) users = [ User(username=f"user{i}", email=f"user{i}@example.com") for i in range(10) ] with get_db() as db: db.add_all(users)

查询数据：

# 获取单个对象 with get_db() as db: user = db.query(User).filter(User.username == "alice").first() # 复杂查询 from sqlalchemy import or_ with get_db() as db: active_users = db.query(User).filter( or_( User.last_login >= datetime.now() - timedelta(days=30), User.posts.any() # 有发过文章的用户 ) ).order_by(User.username).all()

更新数据：

with get_db() as db: user = db.query(User).get(1) # 假设id=1的用户存在 user.email = "new_email@example.com" # 不需要显式调用update，会话会自动跟踪修改

删除数据：

with get_db() as db: user = db.query(User).get(1) db.delete(user) # 或者批量删除 db.query(User).filter(User.username.like("temp%")).delete()

5. 高级查询技巧

5.1 连接查询优化

SQLAlchemy提供了多种连接方式，正确使用能显著提升性能：

# 基本连接(产生N+1查询问题) with get_db() as db: posts = db.query(Post).all() for post in posts: print(post.author.username) # 每次访问author都会产生新查询 # 使用joinedload立即加载关联对象 from sqlalchemy.orm import joinedload with get_db() as db: posts = db.query(Post).options(joinedload(Post.author)).all() for post in posts: print(post.author.username) # 不会产生额外查询

对于多对多关系，可以使用contains_eager：

from sqlalchemy.orm import contains_eager with get_db() as db: posts = db.query(Post).join(Post.tags).filter(Tag.name == "Python").options( contains_eager(Post.tags) ).all()

5.2 聚合与分组

SQLAlchemy支持完整的SQL聚合功能：

from sqlalchemy import func with get_db() as db: # 每个用户的文章数 stats = db.query( User.username, func.count(Post.id).label("post_count") ).join(Post).group_by(User.username).all() # 使用having子句过滤分组结果 active_users = db.query( User.username, func.count(Post.id).label("post_count") ).join(Post).group_by(User.username).having( func.count(Post.id) > 5 ).all()

5.3 子查询与CTE

对于复杂查询，可以使用子查询或CTE(Common Table Expression)：

from sqlalchemy import select # 子查询 subq = select(Post.author_id, func.count('*').label('post_count')).group_by( Post.author_id ).subquery() with get_db() as db: results = db.query( User.username, subq.c.post_count ).join(subq, User.id == subq.c.author_id).all() # CTE (WITH子句) cte = select(Post).where(Post.title.like("%Python%")).cte("python_posts") with get_db() as db: results = db.query(User).join(cte, User.id == cte.c.author_id).all()

6. 事务管理与并发控制

6.1 事务隔离级别

SQLAlchemy支持标准的事务隔离级别：

from sqlalchemy import create_engine engine = create_engine( "postgresql://user:password@localhost/db", isolation_level="REPEATABLE READ" )

不同数据库支持的隔离级别可能不同：

READ UNCOMMITTED (最低隔离，可能读到脏数据)
READ COMMITTED (大多数数据库默认级别)
REPEATABLE READ (避免不可重复读)
SERIALIZABLE (最高隔离，性能影响大)

6.2 处理并发冲突

乐观并发控制是处理并发的常见方式：

from sqlalchemy import Column, Integer, String, DateTime from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker Base = declarative_base() class Product(Base): __tablename__ = 'products' id = Column(Integer, primary_key=True) name = Column(String(50)) quantity = Column(Integer) version_id = Column(Integer, nullable=False) # 用于乐观锁 __mapper_args__ = { "version_id_col": version_id } # 使用示例 with get_db() as db: product = db.query(Product).get(1) product.quantity -= 1 try: db.commit() # 这里会自动检查version_id是否变化 except StaleDataError: db.rollback() print("数据已被其他事务修改，请重试")

7. 性能调优实战

7.1 连接池配置

合理的连接池配置对性能至关重要：

engine = create_engine( "postgresql://user:password@localhost/db", pool_size=5, # 常驻连接数 max_overflow=10, # 允许临时创建的连接数 pool_timeout=30, # 获取连接超时时间(秒) pool_recycle=3600, # 连接回收时间(秒) pool_pre_ping=True # 执行前检查连接是否有效 )

7.2 批量操作优化

批量插入数据时，使用bulk操作能显著提升性能：

# 普通方式(效率低) with get_db() as db: for i in range(1000): db.add(User(username=f"user{i}")) # 批量插入(高效) with get_db() as db: db.bulk_insert_mappings( User, [{"username": f"user{i}"} for i in range(1000)] )

对于更新操作，可以使用bulk_update_mappings：

with get_db() as db: db.bulk_update_mappings( User, [{"id": 1, "username": "new_name1"}, {"id": 2, "username": "new_name2"}] )

8. 常见问题与解决方案

8.1 连接泄漏排查

连接泄漏是常见问题，可以通过事件监听来检测：

from sqlalchemy import event from sqlalchemy.pool import Pool @event.listens_for(Pool, "checkout") def on_checkout(dbapi_conn, connection_record, connection_proxy): print(f"Checkout connection: {id(dbapi_conn)}") @event.listens_for(Pool, "checkin") def on_checkin(dbapi_conn, connection_record): print(f"Checkin connection: {id(dbapi_conn)}")

如果发现checkout数量持续增长而checkin很少，就说明存在连接泄漏。

8.2 N+1查询问题

这是ORM常见性能陷阱：

# 反例(N+1查询) with get_db() as db: users = db.query(User).all() for user in users: print(user.posts) # 每个user.posts都会产生新查询 # 解决方案1: joinedload with get_db() as db: users = db.query(User).options(joinedload(User.posts)).all() # 解决方案2: selectinload (适合一对多) from sqlalchemy.orm import selectinload with get_db() as db: users = db.query(User).options(selectinload(User.posts)).all()

8.3 长事务问题

长时间运行的事务会占用数据库资源并可能导致锁争用：

# 反例 with get_db() as db: user = db.query(User).get(1) # 这里进行耗时操作(如网络请求) time.sleep(30) user.name = "New Name" db.commit() # 事务保持打开状态30秒 # 解决方案: 分离耗时操作 with get_db() as db: user = db.query(User).get(1) db.expunge(user) # 从会话中分离 # 执行耗时操作 time.sleep(30) user.name = "New Name" # 重新附加并提交 with get_db() as db: db.add(user) db.commit()

9. 实际项目经验分享

9.1 多数据库支持策略

在微服务架构中，我们经常需要连接多个数据库。SQLAlchemy可以很好地支持这种场景：

# 多引擎配置 user_engine = create_engine("postgresql://user:pass@user-db/db") order_engine = create_engine("mysql://user:pass@order-db/db") # 绑定不同模型到不同引擎 from sqlalchemy.orm import registry mapper_registry = registry() class User(Base): __tablename__ = 'users' __table_args__ = {'schema': 'auth'} # ... 字段定义 class Order(Base): __tablename__ = 'orders' __table_args__ = {'schema': 'commerce'} # ... 字段定义 # 配置绑定 mapper_registry.configure(binds={ User: user_engine, Order: order_engine })

9.2 数据库迁移管理

对于模型变更，推荐使用Alembic进行迁移：

pip install alembic alembic init migrations

编辑alembic.ini和migrations/env.py后，可以生成迁移脚本：

alembic revision --autogenerate -m "add user table" alembic upgrade head

9.3 测试策略

数据库相关测试应该使用事务回滚：

import pytest from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker @pytest.fixture def db_session(): engine = create_engine("sqlite:///:memory:") Session = sessionmaker(bind=engine) Base.metadata.create_all(engine) session = Session() try: yield session finally: session.rollback() session.close() def test_user_creation(db_session): user = User(username="test", email="test@example.com") db_session.add(user) db_session.flush() assert user.id is not None assert db_session.query(User).count() == 1

10. 扩展与进阶

10.1 混合属性(Hybrid Attributes)

混合属性可以在Python和SQL层面都起作用：

from sqlalchemy.ext.hybrid import hybrid_property class User(Base): # ... 其他字段 first_name = Column(String(50)) last_name = Column(String(50)) @hybrid_property def full_name(self): return f"{self.first_name} {self.last_name}" @full_name.expression def full_name(cls): return cls.first_name + " " + cls.last_name # 使用示例 with get_db() as db: # Python层面 user = db.query(User).first() print(user.full_name) # SQL层面 users = db.query(User).filter(User.full_name == "John Doe").all()

10.2 自定义查询类

通过继承Query类可以添加通用过滤条件：

from sqlalchemy.orm import Query class SoftDeleteQuery(Query): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._filter_deleted = True def include_deleted(self): self._filter_deleted = False return self def _get_entities(self): entities = super()._get_entities() if self._filter_deleted: for entity in entities: if hasattr(entity, 'is_deleted'): self._criterion = self._criterion & (entity.is_deleted == False) return entities # 使用自定义查询类 Session = sessionmaker(query_cls=SoftDeleteQuery)

10.3 事件监听系统

SQLAlchemy的事件系统非常强大，可以实现各种横切关注点：

from sqlalchemy import event @event.listens_for(User, 'before_insert') def before_user_insert(mapper, connection, target): if not target.username: raise ValueError("Username is required") target.created_at = datetime.now() @event.listens_for(Engine, 'connect') def set_sqlite_pragma(dbapi_connection, connection_record): if isinstance(dbapi_connection, sqlite3.Connection): cursor = dbapi_connection.cursor() cursor.execute("PRAGMA foreign_keys=ON") cursor.close()

11. 性能监控与分析

11.1 SQL日志分析

启用详细日志记录有助于性能分析：

import logging logging.basicConfig() logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

更详细的性能分析可以使用sqlalchemy.echo参数：

engine = create_engine("postgresql://user:pass@localhost/db", echo="debug")

11.2 使用性能分析工具

对于复杂查询，可以使用EXPLAIN ANALYZE：

from sqlalchemy import text with get_db() as db: result = db.execute(text("EXPLAIN ANALYZE SELECT * FROM users WHERE username = 'test'")) for row in result: print(row[0])

对于Python层面的性能分析，可以使用cProfile：

import cProfile def run_query(): with get_db() as db: db.query(User).filter(User.username.like("test%")).all() cProfile.runctx('run_query()', globals(), locals())

12. 安全最佳实践

12.1 SQL注入防护

SQLAlchemy的查询构建器自动处理参数化查询，但直接使用文本SQL时仍需小心：

# 安全方式(参数化查询) with get_db() as db: db.query(User).filter(User.username == request.args.get('username')) # 危险方式(字符串拼接) with get_db() as db: db.execute(f"SELECT * FROM users WHERE username = '{request.args.get('username')}'")

12.2 敏感数据处理

对于密码等敏感信息，应该使用加密存储：

from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") class User(Base): # ... 其他字段 _password = Column("password", String(128)) @property def password(self): return self._password @password.setter def password(self, value): self._password = pwd_context.hash(value) def verify_password(self, plain_password): return pwd_context.verify(plain_password, self._password)

12.3 权限最小化

数据库用户应该遵循最小权限原则：

-- 只授予必要的权限 CREATE USER app_user WITH PASSWORD 'secure_password'; GRANT CONNECT ON DATABASE mydb TO app_user; GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO app_user;

13. 实际案例：电商系统数据层设计

让我们通过一个电商系统的例子展示SQLAlchemy在实际项目中的应用：

class Product(Base): __tablename__ = 'products' id = Column(Integer, primary_key=True) sku = Column(String(50), unique=True, nullable=False) name = Column(String(100), nullable=False) description = Column(Text) price = Column(Numeric(10, 2), nullable=False) stock = Column(Integer, default=0) is_active = Column(Boolean, default=True) category_id = Column(Integer, ForeignKey('categories.id')) category = relationship("Category", back_populates="products") variants = relationship("ProductVariant", back_populates="product") images = relationship("ProductImage", order_by="ProductImage.sort_order") class Category(Base): __tablename__ = 'categories' id = Column(Integer, primary_key=True) name = Column(String(50), unique=True, nullable=False) slug = Column(String(50), unique=True, nullable=False) parent_id = Column(Integer, ForeignKey('categories.id')) products = relationship("Product", back_populates="category") children = relationship("Category", backref=backref("parent", remote_side=[id])) class Order(Base): __tablename__ = 'orders' id = Column(Integer, primary_key=True) user_id = Column(Integer, ForeignKey('users.id'), nullable=False) status = Column(String(20), nullable=False, default="pending") total_amount = Column(Numeric(10, 2), nullable=False) created_at = Column(DateTime, server_default=func.now()) user = relationship("User", back_populates="orders") items = relationship("OrderItem", back_populates="order") class OrderItem(Base): __tablename__ = 'order_items' id = Column(Integer, primary_key=True) order_id = Column(Integer, ForeignKey('orders.id'), nullable=False) product_id = Column(Integer, ForeignKey('products.id'), nullable=False) quantity = Column(Integer, nullable=False) unit_price = Column(Numeric(10, 2), nullable=False) order = relationship("Order", back_populates="items") product = relationship("Product")

这个设计考虑了电商系统的核心需求：

商品分类的层级结构
商品变体(如不同颜色、尺寸)
订单状态管理
价格计算的精确性(使用Numeric类型)
关联关系的合理设计

14. 调试技巧与工具

14.1 查看生成的SQL

在开发过程中，查看实际生成的SQL非常有用：

# 方法1: 启用echo engine = create_engine("sqlite:///:memory:", echo=True) # 方法2: 使用str() query = session.query(User).filter(User.username == "test") print(str(query)) # 方法3: 编译查询 from sqlalchemy.dialects import postgresql print(query.statement.compile(dialect=postgresql.dialect()))

14.2 使用SQLAlchemy-Utils

SQLAlchemy-Utils提供了许多有用的扩展：

pip install sqlalchemy-utils

常用功能示例：

from sqlalchemy_utils import ( create_database, drop_database, database_exists, EmailType, PhoneNumberType, ChoiceType ) # 创建/删除数据库 if not database_exists(engine.url): create_database(engine.url) # 特殊字段类型 class User(Base): email = Column(EmailType) phone = Column(PhoneNumberType) status = Column(ChoiceType([ ('active', 'Active'), ('inactive', 'Inactive') ]))

14.3 性能分析插件

使用SQLAlchemy的性能分析插件识别慢查询：

from sqlalchemy import event from sqlalchemy.engine import Engine import time @event.listens_for(Engine, "before_cursor_execute") def before_cursor_execute(conn, cursor, statement, parameters, context, executemany): context._query_start_time = time.time() @event.listens_for(Engine, "after_cursor_execute") def after_cursor_execute(conn, cursor, statement, parameters, context, executemany): duration = time.time() - context._query_start_time if duration > 0.5: # 记录超过0.5秒的查询 print(f"Slow query ({duration:.2f}s): {statement}")

15. 与异步框架集成

15.1 SQLAlchemy与异步IO

SQLAlchemy 1.4+支持异步操作：

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker async_engine = create_async_engine( "postgresql+asyncpg://user:password@localhost/db" ) AsyncSessionLocal = sessionmaker( async_engine, class_=AsyncSession, expire_on_commit=False ) async def get_async_db(): async with AsyncSessionLocal() as db: yield db # 使用示例 async def get_user(user_id: int): async with get_async_db() as db: result = await db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() return user

15.2 在FastAPI中使用

FastAPI与SQLAlchemy的异步集成：

from fastapi import Depends, FastAPI from sqlalchemy.ext.asyncio import AsyncSession app = FastAPI() @app.get("/users/{user_id}") async def read_user(user_id: int, db: AsyncSession = Depends(get_async_db)): result = await db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() return user

16. 版本升级与迁移

16.1 从1.x迁移到2.x

SQLAlchemy 2.0有一些重要变化：

查询API变化：

# 1.x风格 session.query(User).filter(User.name == 'John') # 2.x风格 from sqlalchemy import select session.execute(select(User).where(User.name == 'John'))

Session API变化：

# 1.x风格 session.query(User).get(1) # 2.x风格 session.get(User, 1)

事务管理变化：

# 1.x风格 session.begin() try: # 操作 session.commit() except: session.rollback() raise # 2.x风格 with session.begin(): # 操作

16.2 兼容性策略

为了平稳升级，可以：

先在代码中启用未来标志：

from sqlalchemy import future engine = create_engine("...", future=True) Session = sessionmaker(engine, future=True)

逐步替换旧式API
使用SQLAlchemy 1.4的过渡版本测试
全面测试所有数据库操作

17. 扩展SQLAlchemy功能

17.1 自定义列类型

创建适合业务的自定义类型：

from sqlalchemy import TypeDecorator import json class JSONEncodedDict(TypeDecorator): """将Python字典存储为JSON字符串的列类型""" impl = String def process_bind_param(self, value, dialect): if value is not None: value = json.dumps(value) return value def process_result_value(self, value, dialect): if value is not None: value = json.loads(value) return value # 使用示例 class Product(Base): attributes = Column(JSONEncodedDict)

17.2 多租户支持

实现多租户架构的模式选择策略：

from sqlalchemy import event from sqlalchemy.orm import Session def set_tenant_schema(connection, branch): """设置搜索路径为当前租户的schema""" tenant = get_current_tenant() # 从请求上下文中获取 if tenant: connection.execute(f"SET search_path TO {tenant.schema}, public") # 为引擎和会话添加事件监听 event.listen(Engine, "connect", set_tenant_schema) event.listen(Session, "after_begin", set_tenant_schema)

18. 数据库设计模式

18.1 软删除模式

实现软删除而不实际删除数据：

class SoftDeleteMixin: is_deleted = Column(Boolean, default=False, nullable=False) def delete(self): self.is_deleted = True class User(SoftDeleteMixin, Base): __tablename__ = 'users' # ... 其他字段 # 查询时自动过滤已删除记录 @event.listens_for(Query, "before_compile", retval=True) def no_deleted(query): for desc in query.column_descriptions: if hasattr(desc['type'], 'is_deleted'): query = query.filter(desc['type'].is_deleted == False) return query

18.2 审计日志模式

跟踪数据变更历史：

from sqlalchemy import event from datetime import datetime class AuditLog(Base): __tablename__ = 'audit_logs' id = Column(Integer, primary_key=True) table_name = Column(String(50), nullable=False) record_id = Column(Integer, nullable=False) action = Column(String(10), nullable=False) # CREATE/UPDATE/DELETE changed_data = Column(JSON) changed_at = Column(DateTime, default=datetime.now) changed_by = Column(Integer) def track_changes(mapper, connection, target): """监听模型变更并记录审计日志""" if not target.id: # 新建记录 action = "CREATE" changes = {c.name: getattr(target, c.name) for c in target.__table__.c} else: # 更新记录 action = "UPDATE" changes = { c.name: {"old": target._sa_instance_state.committed_state.get(c.name), "new": getattr(target, c.name)} for c in target.__table__.c if c.name in target._sa_instance_state.committed_state and target._sa_instance_state.committed_state[c.name] != getattr(target, c.name) } if changes: # 如果有变更 audit_log = AuditLog( table_name=target.__tablename__, record_id=target.id, action=action, changed_data=changes, changed_by=get_current_user_id() ) Session.object_session(target).add(audit_log) # 为所有模型注册事件 for model in Base.__subclasses__(): event.listen(model, 'after_insert', track_changes) event.listen(model, 'after_update', track_changes)

19. 性能基准测试

19.1 ORM vs Core性能对比

在某些场景下，SQLAlchemy Core比ORM更高效：

# ORM方式 def test_orm_insert(benchmark): def insert_users(): with Session() as session: for i in range(1000): session.add(User(username=f"user{i}")) session.commit() benchmark(insert_users) # Core方式 def test_core_insert(benchmark): def insert_users(): with engine.connect() as conn: conn.execute( User.__table__.insert(), [{"username": f"user{i}"} for i in range(1000)] ) conn.commit() benchmark(insert_users)

典型结果：

ORM插入1000条记录：约120ms
Core插入1000条记录：约40ms

19.2 批量操作优化效果

批量操作可以显著提升性能：

# 单条插入 def test_single_insert(benchmark): with Session() as session: @benchmark def insert_users(): for i in range(100): session.add(User(username=f"user{i}")) session.commit() # 批量插入 def test_bulk_insert(benchmark): with Session() as session: @benchmark def insert_users(): session.bulk_insert_mappings( User, [{"username": f"user{i}"} for i in range(100)] )