Python SQLAlchemy ORM 核心实战指南

class Tag(Base): tablename = 'tags' id = Column(Integer, primary_key=True, index=True) name = Column(String(30), unique=True, nullable=False) posts = relationship("Post", secondary="post_tags", back_populates="tags")

关联表（用于多对多关系）

class PostTag(Base): tablename = 'post_tags' post_id = Column(Integer, ForeignKey('posts.id'), primary_key=True) tag_id = Column(Integer, ForeignKey('tags.id'), primary_key=True)

创建数据库表

python

创建所有表

Base.metadata.create_all(bind=engine)

删除所有表

Base.metadata.drop_all(bind=engine)

基本 CRUD 操作

创建数据

python

创建新用户

new_user = User(name="张三", email="[email protected]") session.add(new_user) session.commit()

批量创建

session.add_all([ User(name="李四", email="[email protected]"), User(name="王五", email="[email protected]") ]) session.commit()

读取数据

python

获取所有用户

users = session.query(User).all()

获取第一个用户

first_user = session.query(User).first()

根据 ID 获取用户

user = session.query(User).get(1)

更新数据

python

查询并更新

user = session.query(User).get(1) user.name = "张三四" session.commit()

批量更新

session.query(User).filter(User.name.like("张%")).update({"name": "张氏"}, synchronize_session=False) session.commit()

删除数据

python

查询并删除

user = session.query(User).get(1) session.delete(user) session.commit()

批量删除

session.query(User).filter(User.name == "李四").delete(synchronize_session=False) session.commit()

查询数据

基本查询

python

获取所有记录

users = session.query(User).all()

获取特定字段

names = session.query(User.name).all()

排序

users = session.query(User).order_by(User.name.desc()).all()

限制结果数量

users = session.query(User).limit(10).all()

偏移量

users = session.query(User).offset(5).limit(10).all()

过滤查询

python

from sqlalchemy import or_

等值过滤

user = session.query(User).filter(User.name == "张三").first()

模糊查询

users = session.query(User).filter(User.name.like("张%")).all()

IN 查询

users = session.query(User).filter(User.name.in_(["张三", "李四"])).all()

多条件查询

users = session.query(User).filter( User.name == "张三", User.email.like("%@example.com") ).all()

或条件

users = session.query(User).filter( or_(User.name == "张三", User.name == "李四") ).all()

不等于

users = session.query(User).filter(User.name != "张三").all()

聚合查询

python

from sqlalchemy import func

计数

count = session.query(User).count()

分组计数

user_post_count = session.query( User.name, func.count(Post.id) ).join(Post).group_by(User.name).all()

求和、平均值等

avg_id = session.query(func.avg(User.id)).scalar()

连接查询

python

内连接

results = session.query(User, Post).join(Post).filter(Post.title.like("%Python%")).all()

左外连接

results = session.query(User, Post).outerjoin(Post).all()

指定连接条件

results = session.query(User, Post).join(Post, User.id == Post.author_id).all()

关系操作

python

创建带关系的对象

user = User(name="赵六", email="[email protected]") post = Post(title="我的第一篇博客", content="Hello World!", author=user) session.add(post) session.commit()

通过关系访问

print(f"文章 '{post.title}' 的作者是 {post.author.name}") print(f"用户 {user.name} 的所有文章:") for p in user.posts: print(f" - {p.title}")

多对多关系操作

python_tag = Tag(name="Python") sqlalchemy_tag = Tag(name="SQLAlchemy") post.tags.append(python_tag) post.tags.append(sqlalchemy_tag) session.commit()

print(f"文章 '{post.title}' 的标签:") for tag in post.tags: print(f" - {tag.name}")

事务管理

python

自动提交事务

try: user = User(name="测试用户", email="[email protected]") session.add(user) session.commit() except Exception as e: session.rollback() print(f"发生错误：{e}")

使用事务上下文管理器

from sqlalchemy.orm import Session

def create_user(session: Session, name: str, email: str): try: user = User(name=name, email=email) session.add(user) session.commit() return user except: session.rollback() raise

嵌套事务

with session.begin_nested(): user = User(name="事务用户", email="[email protected]") session.add(user) # 保存点 savepoint = session.begin_nested() try: user = User(name="保存点用户", email="[email protected]") session.add(user) savepoint.commit() except: savepoint.rollback()

最佳实践

1. 会话管理：为每个请求创建新会话，请求结束后关闭
2. 异常处理：始终处理异常并适当回滚事务
3. 延迟加载：注意 N+1 查询问题，使用 eager loading 优化
4. 连接池：合理配置连接池大小和超时设置
5. 数据验证：在模型层或应用层验证数据完整性

python

使用上下文管理器管理会话

from contextlib import contextmanager

@contextmanager def get_db(): db = SessionLocal() try: yield db db.commit() except Exception: db.rollback() raise finally: db.close()

使用示例

with get_db() as db: user = User(name="上下文用户", email="[email protected]") db.add(user)

总结

SQLAlchemy ORM 提供了强大而灵活的数据库操作方式，通过本文的介绍，您应该能够：

1. 安装和配置 SQLAlchemy
2. 定义数据模型和关系
3. 执行基本的 CRUD 操作
4. 构建复杂查询
5. 管理数据库事务
6. 遵循最佳实践

SQLAlchemy 还有更多高级特性，如混合属性、事件监听、自定义查询等，值得进一步探索学习。