SQLAlchemy 2.0 高级特性详解
一、关系运算符 any 与 has
1. any 方法
用于一对多关系中，检查集合中是否存在满足条件的元素。
from sqlalchemy import select# 示例：查找有任何价格大于100的商品的订单
stmt = select(Order).where(Order.items.any(Item.price > 100))# 带多个条件
stmt = select(Order).where(Order.items.any(and_(Item.price > 100, Item.status == 'active')
))
2. has 方法
用于多对一关系中，检查关联对象是否满足条件。
# 示例：查找属于特定用户的订单
stmt = select(Order).where(Order.user.has(User.name == 'John'))
二、Session 缓存机制
1. 一级缓存（Session Cache）
from sqlalchemy.orm import Sessionsession = Session()# 第一次查询会访问数据库
user = session.query(User).filter_by(id=1).first()# 第二次查询会直接从缓存获取
same_user = session.query(User).filter_by(id=1).first()
2. 缓存刷新
# 使所有对象过期
session.expire_all()# 使特定对象过期
session.expire(user)# 刷新特定对象
session.refresh(user)
三、Mapped Column 默认值设置
1. default
在 Python 层面设置默认值：
from sqlalchemy.orm import mapped_columnclass User(Base):__tablename__ = 'users'id = mapped_column(Integer, primary_key=True)status = mapped_column(String, default='active')
2. server_default
在数据库层面设置默认值：
from sqlalchemy import textcreated_at = mapped_column(DateTime,server_default=text('CURRENT_TIMESTAMP')
)
3. default_factory
使用函数动态生成默认值：
from datetime import datetimeclass User(Base):created_at = mapped_column(DateTime,default_factory=datetime.utcnow)
四、Join 操作和关系加载
1. 基本 Join
# 简单连接
stmt = select(User).join(User.addresses)# 指定连接条件
stmt = select(User).join(Address, User.id == Address.user_id)
2. Options 加载策略
from sqlalchemy.orm import joinedload, selectinload# 使用 joinedload 预加载关系
stmt = select(User).options(joinedload(User.addresses))# 使用 selectinload 进行分离加载
stmt = select(User).options(selectinload(User.addresses))
五、级联操作（Cascades）
1. 基本级联设置
class User(Base):__tablename__ = 'users'id = mapped_column(Integer, primary_key=True)addresses = relationship("Address",cascade="all, delete-orphan")
2. 常用级联选项
save-update: 默认选项，保存关联对象delete: 删除关联对象delete-orphan: 删除孤立对象merge: 合并关联对象refresh-expire: 刷新或过期关联对象
六、性能优化建议
1. 查询优化
# 使用 select_from 优化复杂查询
stmt = (select(User).select_from(User).join(User.addresses).options(contains_eager(User.addresses))
)
2. 缓存使用
# 合理使用 Session 缓存
session = Session()# 批量操作时清理缓存
if len(results) > 1000:session.expire_all()
3. 延迟加载
class User(Base):addresses = relationship("Address",lazy="dynamic" # 延迟加载)
这些特性和技巧能帮助你更好地使用 SQLAlchemy 2.0，提高应用性能和代码质量。记住要根据具体场景选择合适的特性和优化策略。
