跳转至

第八章:最佳实践

模型设计

使用 Mixin

from datetime import datetime

class TimestampMixin:
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, onupdate=datetime.utcnow)

class User(Base, TimestampMixin):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(50))

使用 repr

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100))

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

    def to_dict(self):
        return {
            'id': self.id,
            'name': self.name,
            'email': self.email
        }

Session 管理

FastAPI 集成

from fastapi import Depends
from sqlalchemy.orm import Session

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get('/users/')
def get_users(db: Session = Depends(get_db)):
    users = db.query(User).all()
    return users

请求级别 Session

from contextlib import contextmanager

@contextmanager
def get_session():
    session = Session()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()

迁移管理

Alembic 配置

pip install alembic
alembic init migrations

生成迁移

# 自动生成迁移
alembic revision --autogenerate -m "Add user table"

# 执行迁移
alembic upgrade head

# 回滚
alembic downgrade -1

常见问题

1. N+1 问题

# 问题
users = session.query(User).all()
for user in users:
    print(user.posts)  # N 次查询

# 解决
users = session.query(User).options(
    selectinload(User.posts)
).all()

2. 内存泄漏

# 问题:Session 未关闭
session = Session()
users = session.query(User).all()
# 忘记 session.close()

# 解决:使用上下文管理器
with Session() as session:
    users = session.query(User).all()

3. 连接池耗尽

# 配置合理的连接池
engine = create_engine(
    DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True
)

小结

最佳实践要点:

  • 模型设计:Mixin、repr、to_dict
  • Session 管理:FastAPI 集成、上下文管理器
  • 迁移管理:Alembic
  • 常见问题:N+1、内存泄漏、连接池

完成本教程后,你应该能够熟练使用 SQLAlchemy 进行数据库操作。