SQLAlchemy 2.0 高级特性详解

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 Session

session = 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_column

class User(Base):
    __tablename__ = 'users'
    
    id = mapped_column(Integer, primary_key=True)
    status = mapped_column(String, default='active')

2. server_default

在数据库层面设置默认值：

from sqlalchemy import text

created_at = mapped_column(
    DateTime,
    server_default=text('CURRENT_TIMESTAMP')
)

3. default_factory

使用函数动态生成默认值：

from datetime import datetime

class 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，提高应用性能和代码质量。记住要根据具体场景选择合适的特性和优化策略。