SQLAlchemy 是 Python 领域最流行的对象关系映射（ORM）框架之一，它提供了高效且灵活的数据库操作方式。通过本文，我们将深入探讨如何使用 SQLAlchemy ORM 进行实际的数据库交互。
安装依赖
首先确保安装核心库及对应数据库的驱动：
pip install sqlalchemy
针对不同数据库，还需安装相应的驱动程序：
# PostgreSQL
pip install psycopg2-binary
# MySQL
pip install mysql-connector-python
# SQLite (Python 标准库已包含，无需额外安装)
核心概念梳理
在使用之前，理解几个关键组件有助于后续开发：
- Engine：数据库连接引擎，负责与数据库通信及管理连接池。
- Session：会话对象，管理所有持久化操作，相当于一个临时的工作单元。
- Model：数据模型类，直接对应数据库中的表结构。
- Query：查询构建器，用于动态生成和执行 SQL 查询。
建立数据库连接
初始化引擎和会话工厂是第一步。这里以 SQLite 为例，PostgreSQL 或 MySQL 只需修改 URL 即可：
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# 创建数据库连接引擎，echo=True 可打印生成的 SQL 语句便于调试
engine = create_engine('sqlite:///example.db', echo=True)
# 创建会话工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 创建会话实例
session = SessionLocal()
定义数据模型
模型类继承自 Base，通过 Column 定义字段属性。以下示例展示了用户、文章及标签之间的多对多关系：
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, declarative_base
# 创建基类
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True, index=True)
name = Column(String(50), nullable=False)
email = Column(String(100), unique=True, index=True)
# 一对多关系：一个用户可以有多篇文章
posts = relationship("Post", back_populates="author")
class Post(Base):
__tablename__ = 'posts'
id = Column(Integer, primary_key=True, index=True)
title = Column(String(100), nullable=False)
content = Column(String(500))
author_id = Column(Integer, ForeignKey('users.id'))
# 多对一关系：一篇文章属于一个作者
author = relationship("User", back_populates="posts")
# 多对多关系：文章可以有多个标签
tags = relationship("Tag", secondary="post_tags", back_populates="posts")
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)
创建数据库表
在首次运行应用时，需要基于模型元数据创建物理表：
# 创建所有表
Base.metadata.create_all(bind=engine)
# 如需重置数据库，可取消注释删除所有表
# Base.metadata.drop_all(bind=engine)
基本 CRUD 操作
创建数据
添加单个对象后需提交事务，批量添加则使用 add_all：
# 创建新用户
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()
读取数据
查询方法灵活，支持链式调用：
# 获取所有用户
users = session.query(User).all()
# 获取第一个用户
first_user = session.query(User).first()
# 根据 ID 获取用户
user = session.query(User).get(1)
更新数据
直接修改对象属性并提交即可，支持条件批量更新：
# 查询并更新
user = session.query(User).get(1)
user.name = "张三四"
session.commit()
# 批量更新
session.query(User).filter(User.name.like("张%"))\
.update({"name": "张氏"}, synchronize_session=False)
session.commit()
删除数据
注意删除操作同样需要提交：
# 查询并删除
user = session.query(User).get(1)
session.delete(user)
session.commit()
# 批量删除
session.query(User).filter(User.name == "李四")\
.delete(synchronize_session=False)
session.commit()
高级查询技巧
过滤与排序
利用 filter、order_by、limit 等构建复杂查询：
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()
聚合与连接
处理统计数据和跨表关联：
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()
# 内连接
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()
关系操作实战
ORM 的强大之处在于自动处理对象间的关联加载：
# 创建带关系的对象
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}")
事务管理与异常处理
数据库操作必须考虑事务的一致性，建议使用上下文管理器或 try-except 块：
# 手动事务控制
try:
user = User(name="测试用户", email="[email protected]")
session.add(user)
session.commit()
except Exception as e:
session.rollback()
print(f"发生错误：{e}")
# 嵌套事务与保存点
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()
最佳实践建议
- 会话管理：为每个请求创建新会话，请求结束后务必关闭，避免资源泄露。
- 异常处理：始终捕获异常并在失败时回滚事务，防止脏数据。
- 延迟加载：注意 N+1 查询问题，必要时使用
joinedload或selectinload优化 eager loading。 - 连接池配置：根据应用负载合理设置连接池大小和超时时间。
- 数据验证：在模型层或应用层双重校验数据完整性。
推荐使用上下文管理器封装会话逻辑：
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 提供了强大而灵活的数据库操作能力。通过掌握上述内容，你已经能够完成从环境搭建、模型定义到复杂查询与事务管理的完整流程。当然，它还支持混合属性、事件监听等更多高级特性，值得在实际项目中持续探索。