[特殊字符]️ SQLAlchemy 暴击指南：把数据库变成 Python 对象

🗄️ SQLAlchemy 暴击指南：把数据库变成 Python 对象

直接拼 SQL 字符串又脏又容易注入；纯手写 SQL 又重复又难维护。SQLAlchemy 把这个矛盾解开了——用 Python 类描述表，用 Python 表达式写查询，它替你生成正确的 SQL。这篇把 Obsidian 里 SQLAlchemy 的卡片和笔记揉成一条线：从"为什么需要 ORM"一直到"异步 + 事务"，每一段代码都能跑、版本都对齐。你来检查，我兜底。

📑 目录

1. 为什么需要 ORM / SQLAlchemy

先说人话：ORM = Object Relational Mapping，对象关系映射。它把"数据库表"映射成"Python 类"，把"一行数据"映射成"一个对象"。于是你不用写 SQL 字符串，而是写 Python。

SQLAlchemy 有两层，初学者最容易混淆，先分清：

# ❌ 裸 SQL 字符串
sql = "SELECT * FROM users " \\
"WHERE name='" + name + "'"
# 拼接 → 注入风险 + 难维护

# ✅ SQLAlchemy ORM
await session.execute(
select(User).where(User.name == name))
# 参数化、安全、可读

💡 一句话定位 SQLAlchemy 是 Python 生态里事实标准的数据库工具包。FastAPI 官方教程用的就是它。学会它，等于打通了"Python 后端怎么存数据"的任督二脉。

2. 四大核心概念

把下面四个词刻进脑子，后面全是基于它们的组合：

🧠 深挖：连接池 Engine 不会每次查询都新建连接，而是维护一个连接池——重复利用连接，避免频繁握手开销。这也是为什么高并发下要用连接池而不是"每次连一次"。（对应 Wiki 卡片《数据库连接池》。）

3. 现代模型定义（2.0 写法）

这是全文最该记牢、也最容易踩版本坑的地方。SQLAlchemy 2.0 推荐使用 DeclarativeBase + Mapped + mapped_column。

# models.py · 2.0 风格
from sqlalchemy import String, Integer, Float
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
pass

class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50))
age: Mapped[int] = mapped_column(Integer, default=18)
score: Mapped[float] = mapped_column(Float)

🚫 版本坑（必看，否则过不了你的检查） 旧教程里常写的 from sqlalchemy.ext.declarative import declarative_base 再 Base = declarative_base()，以及 Column(Integer) 那种写法，在 2.0 里已废弃（deprecated）。新项目请一律用上面 DeclarativeBase + Mapped + mapped_column 的写法。遇到老代码能看懂即可，自己写别再用旧的。

类型怎么写？

4. 连库 & 建表

用 create_engine 建引擎，再 Base.metadata.create_all 按模型建表（开发/演示够用；生产请用 Alembic 做迁移）。

# database.py · 同步
from sqlalchemy import create_engine
from .models import Base

# echo=True 会把生成的 SQL 打到控制台，学习期很有用
engine = create_engine("sqlite:///./demo.db", echo=True)

# 首次建表（已存在则跳过）
Base.metadata.create_all(engine)

💡 连接串速查 不同数据库只是"连接串"不同：sqlite:///./x.db、postgresql+psycopg://user:pwd@localhost/db、mysql+pymysql://user:pwd@localhost/db。换库基本只改这一行。

5. 完整 CRUD

Create 增、Read 查、Update 改、Delete 删。下面是一套能直接跑的同步示例：

# crud.py
from sqlalchemy.orm import Session
from .database import engine
from .models import User

# 增（Create）
with Session(engine) as s:
u = User(name="xushuai", age=20)
s.add(u)
s.commit() # 必须 commit 才真正写入
s.refresh(u) # 把数据库生成的 id 同步回对象
print(u.id) # 此时才有值

# 查（Read）
with Session(engine) as s:
u = s.get(User, 1) # 按主键查，最快
print(u.name)

# 改（Update）
with Session(engine) as s:
u = s.get(User, 1)
u.age = 21 # 改属性即改记录
s.commit()

# 删（Delete）
with Session(engine) as s:
u = s.get(User, 1)
s.delete(u)
s.commit()

⚠️ 最常见的两个坑 ① 忘了 commit()——内存里改了，库里没动。② 忘了 refresh() 就读 id——自增主键是数据库生成的，commit 后还需 refresh 才能拿到。这两个点面试/实操高频出现。

6. 进阶查询

2.0 推荐用 select() 构造语句，再用 session.execute(…) 执行，scalars() 取对象列表。

from sqlalchemy import select, func, or_

# 条件过滤（where 等价于旧版 filter）
stmt = select(User).where(User.age >= 18)
users = s.scalars(stmt).all()

# 或条件
stmt = select(User).where(or_(User.age < 18, User.age > 60))

# 模糊匹配（LIKE %帅%）
stmt = select(User).where(User.name.contains("帅"))
# 或手写 like：User.name.like("%帅%")

# 排序 + 分页
stmt = select(User).order_by(User.age.desc()).offset(0).limit(10)

# 聚合：总数 / 平均年龄
total = s.scalar(select(func.count()).select_from(User))
avg_age = s.scalar(select(func.avg(User.age)))

Join 联表（配合下一节的关系）：

# 查出"xushuai 写的所有文章"
stmt = select(Article).join(User).where(User.name == "xushuai")
articles = s.scalars(stmt).all()

7. 关系映射

表与表之间有关系，SQLAlchemy 用 relationship() + ForeignKey 把它们变成对象间的引用。

一对多：一个用户写多篇文章

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship, Mapped, mapped_column

class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50))
articles: Mapped[list["Article"]] = relationship(back_populates="author")

class Article(Base):
__tablename__ = "articles"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(100))
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
author: Mapped["User"] = relationship(back_populates="articles")

用起来就像操作对象：user.articles 直接拿到他的所有文章，article.author 直接拿到作者。

一对一：一个用户对应一份资料

在"一"的那侧加 uselist=False 即可：

class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
profile: Mapped["Profile"] = relationship(
back_populates="user", uselist=False) # 关键

class Profile(Base):
__tablename__ = "profiles"
id: Mapped[int] = mapped_column(primary_key=True)
bio: Mapped[str] = mapped_column(String(200))
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
user: Mapped["User"] = relationship(back_populates="profile")

8. 事务：要么全成，要么全撤

"事务"保证一组操作原子性：要么全部成功提交，要么出错整体回滚，不会出现"钱扣了但订单没生成"的半吊子状态。

with Session(engine) as s:
try:
s.add(User(name="A"))
s.add(User(name="B"))
s.commit() # 两条一起落库
except Exception:
s.rollback() # 出错 → 全部撤销，库里干干净净
raise

💡 小知识 在 2.0 里，with Session() as s: 这个上下文管理器本身就有"正常退出自动提交、异常退出自动回滚"的能力。上面显式写 try/except + rollback 是为了可读和可控，也是面试里展示"我懂事务"的标准写法。

9. 原生 SQL（兜底手段）

ORM 覆盖 90% 场景，但遇到复杂报表、窗口函数等，直接写 SQL 更省心。用 text() 安全传参（别直接拼字符串）：

from sqlalchemy import text

with engine.connect() as conn:
result = conn.execute(
text("SELECT * FROM users WHERE age > :age"),
{"age": 18}, # 参数化，防注入
)
for row in result:
print(row) # row 是类似元组的对象，可按列名取

10. 异步 SQLAlchemy

高并发接口要用异步版：create_async_engine + async_sessionmaker + AsyncSession。注意数据库驱动也要换异步的（如 PostgreSQL 用 asyncpg，SQLite 用 aiosqlite）。

# database_async.py · 异步
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy import select
from .models import User

# 连接串前缀多了 +aiosqlite / +asyncpg
engine = create_async_engine("sqlite+aiosqlite:///./demo.db")
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_users():
async with AsyncSessionLocal() as session:
result = await session.execute(select(User))
return result.scalars().all()

和 FastAPI 配合时，用 lifespan 在启动时建表，用 yield 依赖把 Session 注入接口：

# main_async.py · FastAPI 集成
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from typing import Annotated

@asynccontextmanager
async def lifespan(app: FastAPI):
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all) # 启动建表
yield # 应用运行期

app = FastAPI(lifespan=lifespan)

async def get_db():
async with AsyncSessionLocal() as session:
yield session # 注入后自动关闭

@app.get("/users/")
async def list_users(db: Annotated[AsyncSession, Depends(get_db)]):
res = await db.execute(select(User))
return res.scalars().all()

🧠 深挖：同步 vs 异步 怎么选？ 学习/小项目用同步（create_engine + Session）最省心。要扛高并发、配合 async def 接口，才上异步。注意异步必须配异步驱动，且 ORM 操作要 await。两篇博客打通后你会发现：FastAPI 管"接口"，SQLAlchemy 管"数据"，两者用 Depends 一接就活了。

自检清单（点开看答案）

Q1：Engine、Session、Base、Model 四者分别是什么角色？

Engine=连接总入口/连接池；Session=一次数据库对话的工作单元（增删改查都在它里）；Base=所有模型的声明式父类；Model=一张表对应一个类。关系：Engine 造 Session，Session 操作 Model 实例，Model 继承自 Base。

Q2：SQLAlchemy 2.0 里定义模型，正确的写法是什么？旧的 `declarative_base()` 还能用吗？

正确写法：class Base(DeclarativeBase): pass，字段用 Mapped[类型] = mapped_column(…)。旧 declarative_base() + Column() 写法在 2.0 中已废弃，能跑但不推荐，新代码别用。

Q3：为什么 `add` 之后还要 `commit`，有时还要 `refresh`？

commit() 才真正把改动写入数据库；不 commit 只是内存里的挂起状态。refresh(obj) 把数据库生成的值（如自增 id）同步回 Python 对象，之后才能真正拿到 obj.id。

Q4：一对多和一对一在 `relationship` 上的区别是什么？

"多"的那一侧就是普通 relationship（如 User.articles 是列表）；"一"的那一侧加 uselist=False（如 User.profile 是单个对象）。两端用 back_populates 互指对方属性名，保持双向同步。

Q5：事务的 `rollback` 解决什么问题？什么时候该用它？

解决"一组操作只成功了一部分"的不一致问题。只要多个写操作必须要么全成、要么全撤（如转账：扣款+入账），就要放进同一个 Session，出错时 rollback() 整体回滚，保证数据原子性。

Q6：异步 SQLAlchemy 相比同步，改了哪几处？

① 引擎换 create_async_engine；② 会话换 async_sessionmaker + AsyncSession；③ 连接串加异步驱动前缀（+aiosqlite / +asyncpg）；④ 所有 ORM 操作用 await（await session.execute(…)）。

📚 资料来源 & 版本核查

• 本篇整理自 Obsidian 知识库 03 – 参考资料/数据库/03.SQLAlchemy学习与使用.md、FastAPI 第 07–09 章，以及 06 – Wiki/数据库/ 卡片（SQLAlchemy Core / ORM / Session / Engine / 连接池 / Declarative Base）。
• 版本基准（已联网核对，2026-08-03）：SQLAlchemy 2.0.x（最新稳定线 2.0.51；2.1 处于 2.1.0b3 beta，未 GA）。代码全部采用 2.0 现代写法。
• 关键准确性说明：declarative_base() 自 2.0 起废弃，改用 DeclarativeBase；2.0 推荐 select() + where() + execute()/scalars() 查询范式。