SQLAlchemy ORM完全指南：Python数据库操作实战
概述
SQLAlchemy是Python生态系统中最强大的ORM（对象关系映射）库，它为开发者提供了高效、灵活的数据库操作方案。本文将系统性地介绍SQLAlchemy ORM的核心用法，帮助您快速掌握Python数据库开发技能。
目录
环境准备
首先需要安装SQLAlchemy核心包：
pip install sqlalchemy根据实际使用的数据库类型，还需要安装对应的驱动包：
# PostgreSQL数据库
pip install psycopg2-binary
# MySQL数据库
pip install mysql-connector-python
# SQLite为Python内置库，无需额外安装核心概念
理解以下四个核心组件是掌握SQLAlchemy的关键：
- Engine：数据库连接引擎，负责与数据库服务器通信
- Session：数据库会话对象，管理所有数据的持久化操作
- Model：数据模型类，映射到数据库中的表结构
- Query：查询构建器，用于构造和执行各类数据库查询
数据库连接
使用create_engine函数建立数据库连接：
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# 连接SQLite数据库
engine = create_engine('sqlite:///demo.db', echo=True)
# 连接PostgreSQL数据库
# engine = create_engine('postgresql://user:pass@localhost:5432/dbname')
# 连接MySQL数据库
# engine = create_engine('mysql+mysqlconnector://user:pass@localhost:3306/dbname')
# 初始化会话工厂
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 创建会话实例
db_session = SessionFactory()数据模型定义
使用declarative_base创建模型基类，并定义数据表结构：
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, declarative_base
# 创建模型基类
Base = declarative_base()
class Author(Base):
"""作者表模型"""
__tablename__ = 'authors'
id = Column(Integer, primary_key=True, index=True)
name = Column(String(50), nullable=False)
email = Column(String(100), unique=True, index=True)
# 一对多关系：作者有多篇文章
articles = relationship("Article", back_populates="writer")
class Article(Base):
"""文章表模型"""
__tablename__ = 'articles'
id = Column(Integer, primary_key=True, index=True)
title = Column(String(100), nullable=False)
body = Column(String(1000))
writer_id = Column(Integer, ForeignKey('authors.id'))
# 多对一关系：文章属于一个作者
writer = relationship("Author", back_populates="articles")
# 多对多关系：文章有多个标签
categories = relationship("Category", secondary="article_categories", back_populates="tagged_articles")
class Category(Base):
"""标签分类表模型"""
__tablename__ = 'categories'
id = Column(Integer, primary_key=True, index=True)
name = Column(String(30), unique=True, nullable=False)
tagged_articles = relationship("Article", secondary="article_categories", back_populates="categories")
# 多对多关联表
class ArticleCategory(Base):
__tablename__ = 'article_categories'
article_id = Column(Integer, ForeignKey('articles.id'), primary_key=True)
category_id = Column(Integer, ForeignKey('categories.id'), primary_key=True)表结构创建
根据模型定义创建数据库表：
# 创建所有定义的表
Base.metadata.create_all(bind=engine)
# 如果需要删除所有表
# Base.metadata.drop_all(bind=engine)基础数据操作
插入数据
# 插入单条记录
new_author = Author(name="王小明", email="wangxm@example.com")
db_session.add(new_author)
db_session.commit()
# 批量插入多条记录
db_session.add_all([
Author(name="李小红", email="lixh@example.com"),
Author(name="赵小刚", email="zhaog@example.com")
])
db_session.commit()查询数据
# 查询所有记录
all_authors = db_session.query(Author).all()
# 获取第一条记录
first_author = db_session.query(Author).first()
# 根据主键查询
author = db_session.query(Author).get(1)更新数据
# 查询后更新
author = db_session.query(Author).get(1)
author.name = "王小明更新"
db_session.commit()
# 批量更新
db_session.query(Author).filter(Author.name.like("王%")).update(
{"name": "王氏家族"},
synchronize_session=False
)
db_session.commit()删除数据
# 查询后删除
author = db_session.query(Author).get(1)
db_session.delete(author)
db_session.commit()
# 条件删除
db_session.query(Author).filter(Author.name == "李小红").delete(
synchronize_session=False
)
db_session.commit()数据查询技巧
基础查询
# 查询所有记录
authors = db_session.query(Author).all()
# 只查询特定字段
names = db_session.query(Author.name).all()
# 排序查询
authors = db_session.query(Author).order_by(Author.name.desc()).all()
# 限制返回数量
authors = db_session.query(Author).limit(10).all()
# 分页查询
authors = db_session.query(Author).offset(5).limit(10).all()条件过滤
from sqlalchemy import or_, and_
# 精确匹配
author = db_session.query(Author).filter(Author.name == "王小明").first()
# 模糊匹配
authors = db_session.query(Author).filter(Author.name.like("王%")).all()
# IN查询
authors = db_session.query(Author).filter(
Author.name.in_(["王小明", "李小红"])
).all()
# 多条件AND查询
authors = db_session.query(Author).filter(
Author.name == "王小明",
Author.email.like("%@example.com")
).all()
# OR查询
authors = db_session.query(Author).filter(
or_(Author.name == "王小明", Author.name == "李小红")
).all()
# 不等于查询
authors = db_session.query(Author).filter(Author.name != "王小明").all()聚合统计
from sqlalchemy import func
# 计数统计
total = db_session.query(Author).count()
# 分组统计
stats = db_session.query(
Author.name,
func.count(Article.id)
).join(Article).group_by(Author.name).all()
# 平均值
avg_id = db_session.query(func.avg(Author.id)).scalar()
# 求和
sum_id = db_session.query(func.sum(Author.id)).scalar()关联查询
# 内连接查询
results = db_session.query(Author, Article).join(Article).filter(
Article.title.like("%Python%")
).all()
# 左外连接
results = db_session.query(Author, Article).outerjoin(Article).all()
# 自定义连接条件
results = db_session.query(Author, Article).join(
Article, Author.id == Article.writer_id
).all()关联关系处理
# 创建带有关联的数据
author = Author(name="孙小七", email="sunxq@example.com")
article = Article(title="我的第一篇文章", body="内容", writer=author)
db_session.add(article)
db_session.commit()
# 通过关系属性访问关联数据
print(f"文章《{article.title}》的作者是：{article.writer.name}")
print(f"作者{author.name}的所有文章：")
for art in author.articles:
print(f" - {art.title}")
# 处理多对多关系
python_cat = Category(name="Python")
django_cat = Category(name="Django")
article.categories.append(python_cat)
article.categories.append(django_cat)
db_session.commit()
print(f"文章《{article.title}》的标签：")
for cat in article.categories:
print(f" - {cat.name}")事务控制
# 手动事务控制
try:
new_author = Author(name="测试作者", email="test@example.com")
db_session.add(new_author)
db_session.commit()
except Exception as error:
db_session.rollback()
print(f"事务失败: {error}")
# 使用上下文管理器的事务封装
from sqlalchemy.orm import Session
def create_author(db: Session, name: str, email: str):
try:
author = Author(name=name, email=email)
db.add(author)
db.commit()
return author
except Exception:
db.rollback()
raise
# 嵌套事务（保存点）
with db_session.begin_nested():
author = Author(name="嵌套事务作者", email="nested@example.com")
db_session.add(author)
# 手动保存点
savepoint = db_session.begin_nested()
try:
author = Author(name="保存点作者", savepoint=db_session)
db_session.add(author)
savepoint.commit()
except:
savepoint.rollback()开发最佳实践
遵循以下最佳实践可以显著提升数据库操作的可靠性：
- 会话管理：每个请求创建独立会话，请求处理完毕后及时关闭
- 异常处理：始终捕获异常并执行回滚操作
- 查询优化：警惕N+1查询问题，使用预加载提升性能
- 连接池配置：合理设置连接池大小和超时参数
- 数据校验：在模型层或业务层进行数据完整性校验
# 会话管理的最佳实践
from contextlib import contextmanager
@contextmanager
def get_database():
"""数据库会话上下文管理器"""
db = SessionFactory()
try:
yield db
db.commit()
except Exception:
db.rollback()
raise
finally:
db.close()
# 使用示例
with get_database() as db:
author = Author(name="上下文作者", email="context@example.com")
db.add(author)总结
通过本文的系统学习，您应该能够：
- 完成SQLAlchemy的环境配置和数据库连接
- 设计并定义数据模型及表间关联
- 实现增删改查等基础数据库操作
- 构建多条件、多表连接的复杂查询
- 正确处理数据库事务和异常
- 遵循最佳实践构建健壮的数据库层
SQLAlchemy还提供了许多高级特性，如混合属性、事件监听器、自定义类型等，等待您在实际项目中进一步探索和应用。
