第七章：性能优化

连接池

配置连接池

engine = create_engine(
    'postgresql://user:pass@localhost/db',
    pool_size=10,          # 连接池大小
    max_overflow=5,        # 最大溢出连接
    pool_timeout=30,       # 获取连接超时
    pool_recycle=3600,     # 连接回收时间
    echo=True              # 打印 SQL
)

查询优化

只查询需要的列

# 不推荐
users = session.query(User).all()

# 推荐
users = session.query(User.name, User.email).all()

批量操作

# 批量插入
users = [User(name=f'User{i}') for i in range(1000)]
session.bulk_save_objects(users)
session.commit()

# 批量更新
session.query(User).filter(
    User.is_active == False
).update({'status': 'inactive'}, synchronize_session=False)

使用索引

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    email = Column(String(100), index=True)  # 索引
    name = Column(String(50))

缓存

查询缓存

from sqlalchemy.orm import lazyload

# 禁用关系加载
user = session.query(User).options(
    lazyload('*')
).filter(User.id == 1).first()

异步 SQLAlchemy

异步引擎

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

engine = create_async_engine(
    'postgresql+asyncpg://user:pass@localhost/db',
    echo=True
)

async_session = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

异步查询

async with async_session() as session:
    result = await session.execute(
        select(User).where(User.id == 1)
    )
    user = result.scalar_one()

小结

性能优化要点：

• 连接池：pool_size、max_overflow
• 查询优化：只查需要的列、批量操作
• 索引：index=True
• 异步：create_async_engine、AsyncSession

下一章我们将学习最佳实践。