使用 SQLAlchemy 操作单表：以 SQLite 用户表为例的完整实战指南
作者：张大鹏
日期：2025年11月
一、前言
在 Python 的数据开发与后端工程中，SQLAlchemy 是最流行、最强大的 ORM 框架之一。
它不仅可以让我们像操作对象一样操作数据库，还能同时支持 ORM 与 原生 SQL 两种方式。
本文将通过一个完整的示例，使用 SQLite 数据库和一张简单的 用户表（User），
详细介绍 SQLAlchemy 的单表操作：包括新增、查询、更新、删除、过滤、排序、分页等核心技术。
二、环境准备
1️⃣ 安装依赖
pip install sqlalchemy
✅ SQLite 为 Python 内置数据库，无需额外安装驱动。
2️⃣ 目录结构
project/
├── main.py
└── users.db
我们将在 main.py 中编写全部逻辑。
三、创建数据库连接与模型
1️⃣ 导入依赖并创建引擎
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker# 创建数据库连接（SQLite）
DATABASE_URL = "sqlite:///users.db"engine = create_engine(DATABASE_URL, echo=True)# 创建基础类
Base = declarative_base()# 创建会话工厂
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
2️⃣ 定义用户模型
class User(Base):__tablename__ = "users"id = Column(Integer, primary_key=True, autoincrement=True)username = Column(String(50), nullable=False, unique=True)email = Column(String(100), nullable=False, unique=True)age = Column(Integer, default=18)def __repr__(self):return f"<User(id={self.id}, username='{self.username}', email='{self.email}', age={self.age})>"
3️⃣ 创建表结构
Base.metadata.create_all(engine)
执行后，将自动在本地生成 users.db 文件，并创建一张 users 表。
四、新增数据（Create）
方式一：单条插入
session = SessionLocal()user = User(username="alice", email="alice@example.com", age=25)
session.add(user)
session.commit()print("✅ 插入成功:", user)
方式二：批量插入
users = [User(username="bob", email="bob@example.com", age=30),User(username="charlie", email="charlie@example.com", age=22),User(username="david", email="david@example.com", age=28),
]
session.add_all(users)
session.commit()print("✅ 批量插入完成")
💡
session.add()用于单条，session.add_all()可添加多条。
插入后必须commit()才会生效。
五、查询数据（Read）
SQLAlchemy 的查询非常灵活。
查询语法基于 session.query()，也可以使用新式 select() 语法（SQLAlchemy 2.x）。
1️⃣ 查询所有用户
users = session.query(User).all()
for u in users:print(u)
2️⃣ 按条件过滤
# 查询单个用户
user = session.query(User).filter(User.username == "alice").first()
print(user)# 查询年龄大于25的用户
users = session.query(User).filter(User.age > 25).all()
print(users)
3️⃣ 模糊匹配与逻辑条件
from sqlalchemy import or_, and_# 模糊查询
users = session.query(User).filter(User.username.like("%a%")).all()# 多条件查询
users = session.query(User).filter(and_(User.age > 20, User.age < 30)
).all()# 或条件
users = session.query(User).filter(or_(User.username == "alice", User.username == "bob")
).all()
4️⃣ 排序与分页
# 按年龄降序
users = session.query(User).order_by(User.age.desc()).all()# 分页：第2页，每页2条
page, page_size = 2, 2
users = session.query(User).offset((page - 1) * page_size).limit(page_size).all()
六、更新数据（Update）
更新操作可通过查询对象再修改属性完成。
1️⃣ 更新单条记录
user = session.query(User).filter(User.username == "alice").first()
if user:user.age = 26session.commit()print("✅ 更新成功:", user)
2️⃣ 批量更新
session.query(User).filter(User.age < 25).update({"age": 25})
session.commit()
print("✅ 批量更新完成")
⚠️ 批量更新不会触发 ORM 对象的事件监听器，适用于直接数据库操作场景。
七、删除数据（Delete）
删除操作同样有两种方式。
1️⃣ 删除单条记录
user = session.query(User).filter(User.username == "bob").first()
if user:session.delete(user)session.commit()print("✅ 已删除:", user)
2️⃣ 批量删除
session.query(User).filter(User.age < 25).delete()
session.commit()
print("✅ 批量删除完成")
八、原生 SQL 查询（可选）
虽然 ORM 更方便，但有时直接执行 SQL 也很有用。
from sqlalchemy import textresult = session.execute(text("SELECT * FROM users WHERE age >= :age"), {"age": 25})
for row in result.mappings():print(row)
输出：
{'id': 1, 'username': 'alice', 'email': 'alice@example.com', 'age': 26}
{'id': 4, 'username': 'david', 'email': 'david@example.com', 'age': 28}
九、事务与异常处理
所有数据库操作都应放在事务中执行，并在异常时回滚。
try:new_user = User(username="eve", email="eve@example.com", age=29)session.add(new_user)session.commit()
except Exception as e:session.rollback()print("❌ 发生错误，已回滚:", e)
finally:session.close()
💡
rollback()可防止部分提交导致数据不一致。
十、总结与最佳实践
| 操作 | 语法 | 说明 |
|---|---|---|
| 新增 | session.add() / session.add_all() | 需 commit() |
| 查询 | session.query(User) | 可配合 filter()、order_by() |
| 更新 | .update({...}) 或对象修改 | 建议单条对象修改 |
| 删除 | .delete() 或 session.delete() | 支持批量 |
| 原生 SQL | session.execute(text()) | 灵活但需注意安全 |
| 事务控制 | try/except/rollback() | 避免脏数据 |
✅ 建议
- 始终在
try/except中进行数据库操作； - 使用上下文管理（
with SessionLocal() as session:）自动关闭会话； - 在生产环境中使用连接池（
pool_pre_ping=True）防止连接断开； - 模型中添加
__repr__方便调试。
十一、完整示例（可直接运行）
from sqlalchemy import create_engine, Column, Integer, String, text, and_
from sqlalchemy.orm import declarative_base, sessionmaker# 数据库连接
engine = create_engine("sqlite:///users.db", echo=False)
Base = declarative_base()
SessionLocal = sessionmaker(bind=engine)# 模型定义
class User(Base):__tablename__ = "users"id = Column(Integer, primary_key=True)username = Column(String(50), nullable=False, unique=True)email = Column(String(100), nullable=False, unique=True)age = Column(Integer)Base.metadata.create_all(engine)# 会话
session = SessionLocal()# 增
session.add_all([User(username="alice", email="alice@example.com", age=25),User(username="bob", email="bob@example.com", age=30),
])
session.commit()# 查
print(session.query(User).filter(User.age > 20).all())# 改
session.query(User).filter(User.username == "alice").update({"age": 26})
session.commit()# 删
session.query(User).filter(User.age < 25).delete()
session.commit()session.close()
十二、延伸阅读
- SQLAlchemy 官方文档
- FastAPI 官方数据库教程
- SQLite 官方介绍
🎯 总结语
通过本文，你已经掌握了 SQLAlchemy 在单表操作中的核心能力：
增、删、改、查、过滤、排序、分页、事务与原生 SQL 执行。
掌握这些知识后，你已经可以：
- 构建一个完整的用户管理系统；
- 进一步学习多表关系（
ForeignKey,relationship）； - 或者迁移到更强大的数据库（如 PostgreSQL、MySQL）。