用 Python 的方式优雅地操作数据库前言在日常 Python 开发中你是否遇到过这些问题写原生 SQL 太繁琐拼接字符串还容易出错换了数据库如从 MySQL 切到 PostgreSQL代码改动量巨大想用一种更“Pythonic”的方式操作数据库如果你的答案是“是”那么SQLAlchemy就是为你而生的工具。SQLAlchemy 是 Python 生态中最流行的 SQL 工具包和对象关系映射器ORM它让你用写 Python 代码的方式优雅地操作数据库。本文将带你从零开始系统掌握 SQLAlchemy 的核心用法。一、什么是 SQLAlchemySQLAlchemy 诞生于 2006 年采用 MIT 许可证支持多种数据库后端。它的架构分为两层Core 层提供 SQL 表达式语言和数据库连接池适合需要精细控制 SQL 的场景ORM 层将数据库表映射为 Python 类支持关系、继承、事务等高级特性小贴士你可以只用 Core 层也可以结合 ORM 层使用灵活切换。二、核心概念速览在使用 SQLAlchemy 之前先搞清楚这几个核心概念概念说明类比Engine数据库连接引擎管理连接池数据库的“总入口”Session数据库会话执行 CRUD 操作一次“通话”Base模型基类所有模型继承它模型的“祖宗”Model映射数据库表的 Python 类表结构的“蓝图”它们的关系是这样的三、环境准备与安装3.1 安装 SQLAlchemybashpip install sqlalchemy如果需要连接特定数据库还需安装对应的驱动bash# PostgreSQL pip install psycopg2-binary # MySQL pip install pymysql # 异步支持SQLAlchemy 2.0 pip install sqlalchemy[asyncio]3.2 验证安装pythonimport sqlalchemy print(sqlalchemy.__version__) # 输出版本号即成功四、第一个 SQLAlchemy 程序让我们从零开始写一个完整的示例。4.1 创建数据库连接Enginepythonfrom sqlalchemy import create_engine # SQLite 内存数据库方便测试 engine create_engine(sqlite:///:memory:, echoTrue) # 也可以使用文件数据库 # engine create_engine(sqlite:///mydb.db) # MySQL 示例 # engine create_engine(mysqlpymysql://user:passwordlocalhost:3306/mydb) # PostgreSQL 示例 # engine create_engine(postgresqlpsycopg2://user:passwordlocalhost/mydb)echoTrue会打印所有执行的 SQL 语句方便调试。4.2 定义模型基类pythonfrom sqlalchemy.orm import DeclarativeBase class Base(DeclarativeBase): pass4.3 定义数据模型SQLAlchemy 2.0 风格SQLAlchemy 2.0 引入了基于Mapped和mapped_column的类型注解方式更加现代且类型安全。pythonfrom typing import Optional, List from sqlalchemy import String, ForeignKey from sqlalchemy.orm import Mapped, mapped_column, relationship class User(Base): __tablename__ users id: Mapped[int] mapped_column(primary_keyTrue) name: Mapped[str] mapped_column(String(50), nullableFalse) email: Mapped[str] mapped_column(String(100), uniqueTrue, nullableFalse) age: Mapped[Optional[int]] mapped_column(defaultNone) # 一对多关系 posts: Mapped[List[Post]] relationship(back_populatesauthor) def __repr__(self) - str: return fUser(id{self.id}, name{self.name}, email{self.email}) class Post(Base): __tablename__ posts id: Mapped[int] mapped_column(primary_keyTrue) title: Mapped[str] mapped_column(String(200), nullableFalse) content: Mapped[str] mapped_column(String(1000)) user_id: Mapped[int] mapped_column(ForeignKey(users.id)) # 多对一关系 author: Mapped[User] relationship(back_populatesposts) def __repr__(self) - str: return fPost(id{self.id}, title{self.title})对比 1.x 风格旧版使用Column和类型作为第一个参数新版使用mapped_column()配合类型注解更加符合 Python 类型提示的最佳实践。4.4 创建表python# 根据所有继承 Base 的模型在数据库中创建对应的表 Base.metadata.create_all(engine)4.5 创建 Sessionpythonfrom sqlalchemy.orm import sessionmaker SessionLocal sessionmaker(bindengine)五、CRUD 操作详解⚠️重要提示从 SQLAlchemy 1.4 开始官方推荐使用 2.0 风格的查询 APIselect()session.query()已被标记为弃用。以下示例均采用新方式。5.1 创建Create- 插入数据pythonwith SessionLocal() as session: # 创建用户对象 new_user User( name张三, emailzhangsanexample.com, age25 ) # 添加到 session session.add(new_user) # 提交事务必须 session.commit() # 刷新后可获取自增 ID print(f新用户 ID: {new_user.id})批量插入pythonwith SessionLocal() as session: users [ User(name李四, emaillisiexample.com, age30), User(name王五, emailwangwuexample.com, age28), ] session.add_all(users) session.commit()⚠️常见错误忘记commit()数据不会真正写入数据库。5.2 查询Read- 读取数据查询所有pythonfrom sqlalchemy import select with SessionLocal() as session: stmt select(User) users session.execute(stmt).scalars().all() for user in users: print(user)条件查询pythonwith SessionLocal() as session: # 查询单个 stmt select(User).where(User.name 张三) user session.execute(stmt).scalars().first() print(user) # 查询多个 stmt select(User).where(User.age 18) users session.execute(stmt).scalars().all()复杂条件pythonfrom sqlalchemy import and_, or_ with SessionLocal() as session: # AND 条件 stmt select(User).where( and_(User.age 18, User.age 60) ) # OR 条件 stmt select(User).where( or_(User.name 张三, User.name 李四) )⚠️易错点忘记.scalars()会得到Row对象而不是模型实例。5.3 更新Update- 修改数据pythonwith SessionLocal() as session: # 先查询再修改 stmt select(User).where(User.name 张三) user session.execute(stmt).scalars().first() if user: user.age 26 session.commit()批量更新更高效pythonfrom sqlalchemy import update with SessionLocal() as session: stmt update(User).where(User.age 18).values(age18) session.execute(stmt) session.commit()5.4 删除Delete- 删除数据pythonwith SessionLocal() as session: # 先查询再删除 stmt select(User).where(User.name 王五) user session.execute(stmt).scalars().first() if user: session.delete(user) session.commit()批量删除pythonfrom sqlalchemy import delete with SessionLocal() as session: stmt delete(User).where(User.age 18) session.execute(stmt) session.commit()六、完整 CRUD 流程图七、关系映射进阶7.1 一对多关系上面的User和Post就是典型的一对多关系一个用户有多篇文章。插入关联数据pythonwith SessionLocal() as session: # 查询用户 stmt select(User).where(User.name 张三) user session.execute(stmt).scalars().first() # 创建文章并关联 post Post( titleSQLAlchemy 入门教程, content这是一篇关于 SQLAlchemy 的教程..., authoruser # 自动关联 user_id ) session.add(post) session.commit()查询关联数据pythonwith SessionLocal() as session: stmt select(User).where(User.name 张三) user session.execute(stmt).scalars().first() # 访问该用户的所有文章 for post in user.posts: print(f{post.title}: {post.content})7.2 多对多关系多对多关系需要一张中间表pythonfrom sqlalchemy import Table, Column, Integer, ForeignKey # 中间表用户-标签 关联表 user_tag Table( user_tag, Base.metadata, Column(user_id, ForeignKey(users.id), primary_keyTrue), Column(tag_id, ForeignKey(tags.id), primary_keyTrue), ) class Tag(Base): __tablename__ tags id: Mapped[int] mapped_column(primary_keyTrue) name: Mapped[str] mapped_column(String(50), uniqueTrue) # 多对多关系 users: Mapped[List[User]] relationship(secondaryuser_tag, back_populatestags) # 在 User 模型中添加 # tags: Mapped[List[Tag]] relationship(secondaryuser_tag, back_populatesusers)八、事务处理8.1 基本事务Session 默认在事务中运行pythonwith SessionLocal() as session: try: user User(name测试, emailtestexample.com) session.add(user) # 其他操作... session.commit() # 全部成功才提交 except Exception as e: session.rollback() # 出错则回滚 print(f事务回滚: {e})8.2 嵌套事务Savepoint使用begin_nested()创建保存点实现部分回滚pythonwith SessionLocal() as session: try: # 外层事务 user1 User(name用户1, emailuser1example.com) session.add(user1) # 内层保存点 with session.begin_nested(): user2 User(name用户2, emailuser2example.com) session.add(user2) # 如果这里出错只回滚到保存点user1 不受影响 session.commit() except Exception as e: session.rollback() print(f全部回滚: {e})九、数据库迁移Alembic在开发过程中表结构会不断变化。手写 SQL 迁移脚本容易出错且难以追踪。Alembic是 SQLAlchemy 官方推荐的迁移工具可以像 Git 管理代码一样版本化管理数据库 Schema。9.1 安装与初始化bashpip install alembic alembic init alembic9.2 配置连接编辑alembic.iniinisqlalchemy.url sqlite:///./mydb.db编辑alembic/env.py导入你的模型pythonfrom myapp.models import Base target_metadata Base.metadata9.3 生成迁移脚本bash# 自动检测模型变化并生成迁移脚本 alembic revision --autogenerate -m 添加用户表9.4 执行迁移bash# 升级到最新版本 alembic upgrade head # 回滚到上一个版本 alembic downgrade -1十、SQLAlchemy 2.0 新特性一览特性说明PEP 484 类型支持使用Mapped和mapped_column实现完全类型注解异步支持AsyncSession和create_async_engine支持 asyncio统一查询 API废弃session.query()统一使用select()更好的类型安全编辑器可获得完整的类型提示和自动补全异步使用示例pythonfrom sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker from sqlalchemy import select async_engine create_async_engine(sqliteaiosqlite:///mydb.db) AsyncSessionLocal async_sessionmaker(async_engine, expire_on_commitFalse) async def get_users(): async with AsyncSessionLocal() as session: stmt select(User) result await session.execute(stmt) return result.scalars().all()十一、常见问题与最佳实践11.1 常见 Bug 及解决方案问题原因解决方案数据插入后查不到忘记commit()确保调用了session.commit()查询返回 Row 而非对象忘记.scalars()使用.scalars().first()或.scalars().all()程序运行一段时间报连接超时会话未关闭导致连接泄漏使用with语句自动管理 Session关系属性访问报错未在关系中使用back_populates双向关系需配置back_populates11.2 最佳实践始终使用with管理 Session自动提交/回滚和关闭避免连接泄漏模型定义使用 2.0 风格Mappedmapped_column获得类型提示支持生产环境使用 Alembic不要用create_all()管理生产数据库善用echoTrue调试开发阶段开启查看生成的 SQL复杂查询使用 Core 层ORM 无法高效表达时直接使用 SQL 表达式总结SQLAlchemy 是 Python 生态中最强大、最灵活的数据库工具包。本文涵盖了✅ SQLAlchemy 的核心架构与概念✅ Engine、Session、Model 的创建与配置✅ 完整的 CRUD 操作2.0 风格 API✅ 一对多、多对多关系映射✅ 事务与回滚处理✅ Alembic 数据库迁移✅ SQLAlchemy 2.0 新特性异步 类型注解掌握 SQLAlchemy你的 Python 开发效率将大幅提升数据库操作也将变得优雅而高效。