第八章:最佳实践¶
模型设计¶
使用 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 配置¶
生成迁移¶
# 自动生成迁移
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 进行数据库操作。