前言在日常 Python 后端开发中处理数据库操作往往是最繁琐的环节之一。无论是拼接原生 SQL 语句时的字符串地狱还是不同数据库之间 SQL 方言的细微差异带来的跨库迁移成本都让开发者头疼不已。代码中充斥着大量难以维护的 SQL 字符串不仅可读性差还容易因为字段名拼写错误埋下线上隐患。SQLAlchemy发音S-Q-L Alchemy 或 sequel alchemy就是为解决这些痛点而生的。它是 Python 生态中最强大、最成熟的对象关系映射Object-Relational Mapping简称 ORM工具库为 Python 开发者提供了与数据库交互的优雅方式。ORM 的核心思想通俗来讲就是用操作 Python 对象的方式操作数据库表。你定义了一个 Python 类这个类就对应数据库中的一张表类的属性就对应表中的字段类的一个实例对象就对应表中的一行数据。从此增删改查数据库不再需要手写 SQL而是调用对象的方法和属性即可。通过本文你将收获从零快速上手 SQLAlchemy、掌握 Engine 引擎和 Session 会话等核心组件、精通全套 CRUD 操作、并能够无缝适配 FastAPI 项目开发。一、SQLAlchemy 核心介绍与优势SQLAlchemy 是 Python 中最具影响力的数据库工具包它为开发者提供了一整套与关系型数据库交互的解决方案。在 Python 数据库开发生态中SQLAlchemy 几乎已经成为事实标准无论是小型 Web 应用还是大型企业级项目都能看到它的身影。从架构上看SQLAlchemy 由两大核心模块组成Core 底层引擎提供了数据库连接池管理、SQL 表达式语言、结果集处理、元数据管理以及类型系统等基础设施。即使不使用 ORM你也可以利用 Core 层构建高性能的数据库操作。ORM 上层映射在 Core 之上构建的对象关系映射层实现了 Python 类与数据库表之间的双向映射让开发者可以用面向对象的思维来操作关系型数据。SQLAlchemy 的核心优势体现在多个方面跨数据库兼容采用统一的 Python 接口操作 MySQL、PostgreSQL、SQLite、Oracle 等主流数据库切换数据库只需修改连接字符串业务代码几乎零改动。高解耦设计数据库操作逻辑与业务逻辑分离模型定义与数据库引擎解耦代码结构清晰易于测试和维护。自动类型转换Python 数据类型与 SQL 数据类型自动映射转换无需手动处理类型匹配问题。事务支持内置完善的事务管理机制支持自动提交、手动提交、回滚等操作保证数据一致性。易于维护相比于硬编码的原生 SQL 字符串ORM 方式的代码更加直观重构和调试也更加方便。二、环境搭建与依赖安装开始使用 SQLAlchemy 之前需要先完成核心库和对应数据库驱动的安装。以下是常见的安装命令# 安装 SQLAlchemy 核心库 pip install sqlalchemy MySQL 驱动推荐使用 PyMySQL 或 mysqlclient pip install pymysql 或 pip install mysqlclient PostgreSQL 驱动 pip install psycopg2-binary SQLite 驱动Python 内置无需额外安装安装完成后可以通过以下代码验证环境是否正常import sqlalchemy print(sqlalchemy.__version__) # 打印版本号确认安装成功重要前置准备在使用 SQLAlchemy 之前需要手动创建好数据库。SQLAlchemy 只会自动创建数据表不会自动创建数据库本身。以 MySQL 为例你需要先登录 MySQL 命令行执行CREATE DATABASE my_database CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;三、SQLAlchemy 五大核心组件重点SQLAlchemy 有五个核心组件理解它们各自的作用和关系是掌握 SQLAlchemy 的关键。1. Engine 引擎Engine 是数据库连接的核心负责管理数据库连接池、执行原始 SQL 语句并可以配置 SQL 日志开关便于调试。创建 Engine 需要使用create_engine()函数传入数据库连接字符串。from sqlalchemy import create_engine MySQL 连接字符串格式mysqlpymysql://用户名:密码主机:端口/数据库名 engine create_engine( mysqlpymysql://root:passwordlocalhost:3306/my_database, echoTrue, # 开启 SQL 日志方便调试 pool_size10, # 连接池大小 max_overflow20 # 最大溢出连接数 )2. Session 会话Session 是数据库交互的桥梁所有 CRUD 操作都需要通过 Session 来完成。它负责将对象的增删改操作暂存到内存中等到合适时机再一次性提交到数据库是实现事务管理的核心载体。3. Base 基类Base 是所有数据表模型的父类。所有需要映射到数据库表的 Python 类都必须继承自这个 Base 基类。Base 由declarative_base()函数创建内部维护了一个元数据注册表记录了所有继承它的模型类及其对应的表结构信息。from sqlalchemy.orm import declarative_base Base declarative_base()4. Model 模型Model 模型是数据表的 Python 表现形式。一个 Python 类对应一张数据库表类的属性对应表中的字段而类的一个实例对象就对应表中的一行数据。这种映射关系让开发者可以用操作普通 Python 对象的方式来操作数据库记录。5. Column 字段Column 用于定义模型类的属性对应数据库表中的哪个字段可以设置字段的类型如 Integer、String、DateTime 等以及各种约束如主键、非空、唯一、默认值等。它是连接 Python 对象属性和数据库表字段的桥梁。四、ORM 核心原理详解ORM对象关系映射的本质是建立 Python 对象与关系型数据库表之间的双向映射通道。可以这样理解数据库表是一个二维网格由行和列组成而 Python 中天然适合表达这种结构的载体就是类和实例。ORM 把表名映射为类名把列名映射为属性名把每一行数据映射为一个实例对象。当你在代码中修改对象属性并调用session.commit()时SQLAlchemy 会在背后自动生成对应的 UPDATE SQL 语句并执行。ORM 带来的五大核心优势开发效率提升不用手写 SQL代码量大幅减少开发速度更快。代码可读性强操作对象的语法比拼接 SQL 字符串直观得多代码即文档。数据库无关性切换底层数据库只需要改一行连接字符串业务代码无需改动。自动防止 SQL 注入ORM 内部使用参数化查询从根本上避免了 SQL 注入风险。易于维护和重构字段改名、表结构调整时只需修改模型定义IDE 可以帮助定位所有引用。下面是一段简单的对比直观感受原生 SQL 和 ORM 开发方式的差异# 原生 SQL 方式需要手动拼接 SQL 字符串 cursor.execute(SELECT * FROM users WHERE age %s, (18,)) rows cursor.fetchall() #ORM 方式像操作普通 Python 对象一样查询 users session.query(User).filter(User.age 18).all()五、数据表模型定义实战先创建 Base 基类这是所有模型的基础from sqlalchemy.orm import declarative_base from sqlalchemy import create_engine engine create_engine(mysqlpymysql://root:passwordlocalhost:3306/my_database) Base declarative_base()接下来定义两个完整的模型——User用户表和 Account账户表展示一对多的关系from sqlalchemy import Column, Integer, String, DateTime, Boolean, Text, ForeignKey from sqlalchemy.orm import relationship from datetime import datetime class User(Base): 用户表模型 tablename users # 指定数据库表名 id Column(Integer, primary_keyTrue, autoincrementTrue, comment用户ID) username Column(String(50), uniqueTrue, nullableFalse, comment用户名) email Column(String(100), uniqueTrue, nullableFalse, comment邮箱) hashed_password Column(String(255), nullableFalse, comment加密密码) is_active Column(Boolean, defaultTrue, comment是否激活) bio Column(Text, nullableTrue, comment个人简介) created_at Column(DateTime, defaultdatetime.now, comment创建时间) updated_at Column(DateTime, defaultdatetime.now, onupdatedatetime.now, comment更新时间) #与 Account 的一对多关系 accounts relationship(Account, back_populatesowner) def repr(self): 返回对象的官方字符串表示主要用于调试 return fUser(id{self.id}, username{self.username}) def str(self): 返回用户友好的字符串表示 return f用户{self.username}{self.email} class Account(Base): 账户表模型 tablename accounts id Column(Integer, primary_keyTrue, autoincrementTrue, comment账户ID) user_id Column(Integer, ForeignKey(users.id), nullableFalse, comment所属用户ID) account_type Column(String(20), defaultsavings, comment账户类型) balance Column(Integer, default0, comment余额单位分) created_at Column(DateTime, defaultdatetime.now, comment创建时间) #反向关系 owner relationship(User, back_populatesaccounts) def repr(self): return fAccount(id{self.id}, type{self.account_type}, balance{self.balance})/code/preSQLAlchemy 提供了丰富的字段类型常用的包括Integer整型String(size)变长字符串需指定最大长度Text长文本类型Boolean布尔值DateTime日期时间Date日期Float浮点数DECIMAL精确十进制数Enum枚举类型LargeBinary二进制大数据常用的字段约束如下primary_keyTrue设置为主键autoincrementTrue自动递增通常配合主键使用uniqueTrue唯一约束nullableFalse不允许为空default值设置默认值indexTrue创建索引comment说明字段注释关于 repr 和 str 的区别repr 返回的是对象的官方字符串表示通常用于调试和开发阶段格式上应尽量明确对象类型和关键属性str 返回用户友好的信息主要用于展示给终端用户。在交互式环境中直接输入对象名时调用的是 repr而 print() 输出时优先调用 str。在项目中建议至少实现 repr方便调试时快速了解对象状态。六、数据表创建与初始化Base.metadata.create_all()是数据表创建的入口方法。它会扫描所有继承了 Base 的模型类读取其 tablename 和 Column 定义然后在数据库中生成对应的 CREATE TABLE 语句并执行。创建所有已注册模型对应的数据表Base.metadata.create_all(engine)这个方法具有一个很重要的特性如果表已经存在则不会重复创建也不会修改已有表结构。这意味着你可以放心地在应用启动脚本中调用它而不用担心覆盖已有数据。但这也意味着如果模型字段发生了变更你需要通过数据库迁移工具如 Alembic来同步表结构。在实际项目中推荐将 Engine、Session 和 Base 统一在一个配置模块中创建和管理database.py —— 数据库统一配置模块from sqlalchemy import create_engine from sqlalchemy.orm import declarative_base, sessionmaker DATABASE_URL mysqlpymysql://root:passwordlocalhost:3306/my_database engine create_engine(DATABASE_URL, echoFalse, pool_size10, max_overflow20) SessionLocal sessionmaker(bindengine, autocommitFalse, autoflushFalse) Base declarative_base()七、Session 会话机制详解Session 会话是操作数据库的入口。创建 Session 工厂的方式如下from sqlalchemy.orm import sessionmaker SessionLocal sessionmaker(bindengine)常规写法手动管理 session 生命周期session SessionLocal() try: #执行数据库操作... session.commit() except Exception: session.rollback() raise finally: session.close()更推荐的做法是使用 with 上下文管理器它会自动处理事务提交和资源回收from contextlib import contextmanager contextmanager def get_session(): 获取数据库会话的上下文管理器 session SessionLocal() try: yield session session.commit() # 正常完成时自动提交 except Exception: session.rollback() # 异常时自动回滚 raise finally: session.close() # 最终关闭会话归还连接池 #使用示例 with get_session() as session: user session.query(User).filter(User.id 1).first() user.bio 更新后的简介事务机制的核心流程当你通过 session.add(obj) 添加对象时这个对象只是被暂存在 Session 的内存空间中并没有真正写入数据库。只有当你调用 session.commit() 时SQLAlchemy 才会将积累的所有变更生成对应的 SQL 语句提交到数据库执行。如果发生异常可以调用 session.rollback() 回滚所有未提交的变更。八、全套 CRUD 实战核心重点1.Create 新增数据from backend.tests.sqlalchemy.utils.sqlalchemy_config import SessionFactory from backend.tests.sqlalchemy.utils.base_model import User with SessionFactory() as session: #单条新增 u1 User(usernamelisi2, password123456) session.add(u1) #多条 add_all u2 User(usernamewangwu2, password654321) session.add_all([u1, u2]) #字典批量插入不用构造对象 user_list [ {username:sunqi,password:123123}, {username:zhouba,password:456456} ] session.bulk_insert_mappings(User, user_list) #SQLAlchemy2.0 insert语法 from sqlalchemy import insert stmt insert(User).values([ {username:zhengshi,password:aaa}, {username:chenshi,password:bbb} ]) session.execute(stmt) session.commit()2.Read 查询数据from backend.tests.sqlalchemy.utils.sqlalchemy_config import SessionFactory from backend.tests.sqlalchemy.utils.base_model import User from sqlalchemy import and_, or_ with SessionFactory() as session: # 查询全部 all_user session.query(User).all() # 根据主键查询不存在返回None user session.get(User,1) #取第一条 first_user session.query(User).first() #只查询部分字段返回元组 res session.query(User.username, User.password).all() #条件过滤 filter u session.query(User).filter(User.username admin).first() #不等于 session.query(User).filter(User.username ! admin).all() #模糊匹配 like session.query(User).filter(User.username.like(%a%)).all() #in 匹配 session.query(User).filter(User.id.in_([1,2,3])).all() #大于小于 session.query(User).filter(User.id 1).all() #and_ 多条件同时成立 session.query(User).filter(and_(User.id1, User.username.like(w%))).all() #or_满足其一即可 session.query(User).filter(or_(User.id 1, User.username.like(w%))).all() #排序 order_by session.query(User).order_by(User.id.desc()).all() #分页 limit offset page_size 2 page_num 2 offset_num (page_num -1)* page_size page_data session.query(User).order_by(User.id).limit(page_size).offset(offset_num).all() #统计数量 total session.query(User).filter(User.username.like(z%)).count()3.Update 更新数据from backend.tests.sqlalchemy.utils.sqlalchemy_config import SessionFactory from backend.tests.sqlalchemy.utils.base_model import User from sqlalchemy import update with SessionFactory() as session: #单条更新先查询修改对象属性 user session.query(User).filter(User.username zhangsan).first() if user: user.password new_password123 session.commit() #批量更新 session.query(User).filter(User.username.like(w%)).update({password:common_password}) session.commit() #2.0 update语法 stmt update(User).where(User.username.like(w%)).values(passwordsqlalchemy2.0_password) session.execute(stmt) session.commit()4.Delete 删除数据from backend.tests.sqlalchemy.utils.sqlalchemy_config import SessionFactory from backend.tests.sqlalchemy.utils.base_model import User from sqlalchemy import delete with SessionFactory() as session: #单条删除 user session.query(User).filter(User.username wangwu).first() if user: session.delete(user) session.commit() #批量删除 session.query(User).filter(User.id3).delete() session.commit() #2.0 delete语法 stmt delete(User).where(User.id18) session.execute(stmt) session.commit()九、高级查询分组与聚合查询SQLAlchemy 提供了 func 模块来使用 SQL 聚合函数常用的包括func.count()计数func.sum()求和func.avg()求平均值func.max()最大值func.min()最小值from sqlalchemy import func # 1. 统计每个账户类型的总数、总余额、平均余额 with get_session() as session: results ( session.query( Account.account_type, func.count(Account.id).label(total_count), func.sum(Account.balance).label(total_balance), func.avg(Account.balance).label(avg_balance), ) .group_by(Account.account_type) .all() ) for row in results: print( f类型{row.account_type}, 数量{row.total_count}, f总余额{row.total_balance}, 平均余额{row.avg_balance} ) # 2. 使用 HAVING 做分组后筛选只取分组总余额大于100000的类型 with get_session() as session: results ( session.query( Account.account_type, func.sum(Account.balance).label(total_balance) ) .group_by(Account.account_type) .having(func.sum(Account.balance) 100000) .all() ) for row in results: print(row.account_type, row.total_balance) # 3. 查询返回格式元组 和 字典转换 with get_session() as session: # 默认返回元组形式只查询部分字段 results_tuple session.query(User.username, User.email).all() for item in results_tuple: # item 是元组: (username, email) print(item[0], item[1]) # 查询完整ORM对象手动转为字典 results_obj session.query(User).all() users_dict [ {id: u.id, username: u.username, email: u.email} for u in results_obj ] print(users_dict)十、高频踩坑总结❗SQLAlchemy 只建表数据库需要手动提前创建不会自动生成数据库。❗做新增、修改、删除一定要执行session.commit()否则数据不会落库。❗尽量用with上下文管理器自动释放连接防止连接池耗尽。❗区分defaultPython 层默认和server_default数据库层默认值。❗批量操作bulk_insert_mappings不会触发模型的 default适合高性能批量导入。十一、项目规范总结SQLAlchemy ORM 流程创建 Engine → 创建 Base 基类 → 定义 Model 模型 → 创建数据表 → 获取 Session 会话 → CRUD 操作 → commit 提交。支持传统session.query()也支持 SQLAlchemy2.0select/update/delete新式 API。ORM 不是完全抛弃 SQL复杂场景依然可以直接执行原生 SQL。是 FastAPI 项目最主流数据库方案RAG 后端项目中用来存储文档、知识库、用户数据。掌握 SQLAlchemy是 Python 后端开发必备技能能极大提升数据库层的开发效率。