Python使用SQLAlchemy操作单表的完整实战指南

目录

• 一、前言
• 二、环境准备
• 1️⃣ 安装依赖
• 2️⃣ 目录结构
• 三、创建数据库连接与模型
• 1️⃣ 导入依赖并创建引擎
• 2️⃣ 定义用户模型
• 3️⃣ 创建表结构
• 四、新增数据（Create）
• 方式一：单条插入
• 方式二：批量插入
• 五、查询数据（Read）
• 1️⃣ 查询所有用户
• 2️⃣ 按条件过滤
• 3️⃣ 模糊匹配与逻辑条件
• 4️⃣ 排序与分页
• 六、更新数据（Update）
• 1️⃣ 更新单条记录
• 2️⃣ 批量更新
• 七、删除数据（Delete）
• 1️⃣ 删除单条记录
• 2️⃣ 批量删除
• 八、原生 SQL 查询（可选）
• 九、事务与异常处理
• 十、总结与最佳实践
• 建议
• 十一、完整示例（可直接运行）
• 总结语

在 Python 的数据开发与后端工程中，SQLAlchemy 是最流行、最强大的 ORM 框架之一。 它不仅可以让我们像操作对象一样操作数据库，还能同时支持 ORM 与 原生 SQL 两种方式。

本文将通过一个完整的示例，使用 SQLite 数据库和一张简单的 用户表（User）， 详细介绍 SQLAlchemy 的单表操作：包括新增、查询、更新、删除、过滤、排序、分页等核心技术。

pip install sqlalchemy

✅ SQLite 为 Python 内置数据库，无需额外安装驱动。

project/
├── main.py
└── users.db

我们将在 main.py 中编写全部逻辑。

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)

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})>"

Base.metadata.create_all(engine)

执行后，将自动在本地生成 users.db 文件，并创建一张 users 表。

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() 才会生效。

SQLAlchemy 的查询非常灵活。 查询语法基于 session.query()，也可以使用新式 select() 语法（SQLAlchemy 2.x）。

users = session.query(User).all()
for u in users:
    print(u)

# 查询单个用户
user = session.query(User).filter(User.username == "alice").first()
print(user)

# 查询年龄大于25的用户
users = session.query(User).filter(User.age > 25).all()
print(users)

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()

# 按年龄降序
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()

更新操作可通过查询对象再修改属性完成。

user = session.query(User).filter(User.username == "alice").first()
if user:
    user.age = 26
    session.commit()
    print("✅ 更新成功:", user)

session.query(User).filter(User.age < 25).update({"age": 25})
session.commit()
print("✅ 批量更新完成")

批量更新不会触发 ORM 对象的事件监听器，适用于直接数据库操作场景。

删除操作同样有两种方式。

user = session.query(User).filter(User.username == "bob").first()
if user:
    session.delete(user)
    session.commit()
    print("✅ 已删除:", user)

session.query(User).filter(User.age < 25).delete()
session.commit()
print("✅ 批量删除完成")

虽然 ORM 更方便，但有时直接执行 SQL 也很有用。

from sqlalchemy import text

result = 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 在单表操作中的核心能力： 增、删、改、查、过滤、排序、分页、事务与原生 SQL 执行。

掌握这些知识后，你已经可以：

• 构建一个完整的用户管理系统；
• 进一步学习多表关系（ForeignKey, relationship）；
• 或者迁移到更强大的数据库（如 PostgreSQL、MySQL）。

以上就是Python使用SQLAlchemy操作单表的完整实战指南的详细内容，更多关于Python使用SQLAlchemy操作单表的资料请关注风君子博客其它相关文章！

您可能感兴趣的文章:

• Python库SQLAlchemy的简介、用法和安装步骤详解
• Python ORM神器之SQLAlchemy基本使用完全指南
• 使用Python和SQLAlchemy实现高效的邮件发送系统
• Python使用SQLAlchemy连接ClickHouse数据库
• python SQLAlchemy 数据库连接池的实现