SQLAlchemy 2.0：Python 最强大的 ORM，这次终于不”重”了
写过 Python Web 的人应该都听过 SQLAlchemy——Python 生态里最成熟、功能最全的 ORM。但很多人被它的学习曲线劝退了，觉得”太重”。
2.0 版本不太一样了：查询语法统一了、异步原生支持了、类型提示完整了。它不再是那个”功能强大但难上手”的框架，而是一个现代 Python 的标配工具。
SQLAlchemy 是什么？解决什么问题？
写 Python 操作数据库，最原始的方式是拼 SQL 字符串：
cursor.execute(f"SELECT * FROM users WHERE age > {age}")
拼字符串容易出事——SQL 注入、类型错误、不同数据库语法不一样。
ORM（对象关系映射）把”数据库表”映射成”Python 类”，你操作对象就是在操作数据库。SQLAlchemy 是 Python 里最成熟的 ORM。
SQLAlchemy 2.0 的三个核心变化：
- 查询统一了：1.x 里 Core 和 ORM 两套 API，2.0 统一用
select()
- 异步原生支持：
async/await 操作数据库，配合 FastAPI 很顺畅
- 类型提示完善：
Mapped + mapped_column，IDE 能帮你自动补全
准备工作：安装和连接
pip install sqlalchemy
pip install asyncpg
pip install aiosqlite
同步连接（简单场景）：
from sqlalchemy import create_engine
engine = create_engine("sqlite:///./app.db", echo=True)
异步连接（生产推荐）：
from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine(
"postgresql+asyncpg://user:password@localhost:5432/mydb",
echo=False,
pool_size=10,
max_overflow=20,
)
echo=True 会在控制台打印 SQL 语句，开发阶段方便调试，生产环境关掉。
定义模型：一张表就是一个类
from datetime import datetime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Boolean, func
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
email: Mapped[str | None] = mapped_column(String(100), nullable=True)
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
created_at: Mapped[datetime] = mapped_column(server_default=func.now())
关键点：
__tablename__ 指定数据库表名Mapped[T] 声明字段类型，IDE 能识别mapped_column() 定义列属性（类型、长度、唯一性、默认值）server_default=func.now() 表示默认值由数据库生成，不是 Python 生成
创建表：
async def init_db():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
CRUD：增删改查
准备工作：创建会话工厂
from sqlalchemy.ext.asyncio import async_sessionmaker
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
async def get_db():
async with AsyncSessionLocal() as session:
yield session
创建（Create）：
async def create_user(db, username: str, email: str = None):
user = User(username=username, email=email)
db.add(user)
await db.commit()
await db.refresh(user)
return user
refresh 让对象从数据库重新加载一遍，拿到 id 这种自动生成的值。
批量创建：
async def create_users(db, users_data: list):
users = [User(**data) for data in users_data]
db.add_all(users)
await db.commit()
return users
查询（Read）：
from sqlalchemy import select
async def get_user_by_id(db, user_id: int):
stmt = select(User).where(User.id == user_id)
result = await db.execute(stmt)
return result.scalar_one_or_none()
async def get_active_users(db, limit: int = 100):
stmt = select(User).where(User.is_active == True).limit(limit)
result = await db.execute(stmt)
return result.scalars().all()
更新（Update）：
async def update_user_email(db, user_id: int, new_email: str):
user = await get_user_by_id(db, user_id)
if user:
user.email = new_email
await db.commit()
await db.refresh(user)
return user
删除（Delete）：
async def delete_user(db, user_id: int) -> bool:
user = await get_user_by_id(db, user_id)
if user:
await db.delete(user)
await db.commit()
return True
return False
关系映射：表之间怎么关联
一对多：一个用户有多篇文章
class User(Base):
__tablename__ = "users"
posts: Mapped[list["Post"]] = relationship(
back_populates="author",
lazy="selectin"
)
class Post(Base):
__tablename__ = "posts"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(200))
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
author: Mapped["User"] = relationship(back_populates="posts")
lazy="selectin" 的意思是：查询用户的时候，顺便把关联的文章也查出来。不加的话，访问 user.posts 时会再触发一次查询（这就是 N+1 问题）。
多对多：用户和角色
from sqlalchemy import Table, Column, Integer, ForeignKey
user_role_table = Table(
"user_roles",
Base.metadata,
Column("user_id", Integer, ForeignKey("users.id")),
Column("role_id", Integer, ForeignKey("roles.id")),
)
class Role(Base):
__tablename__ = "roles"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50), unique=True)
users: Mapped[list["User"]] = relationship(
secondary=user_role_table,
back_populates="roles"
)
roles: Mapped[list["Role"]] = relationship(
secondary=user_role_table,
back_populates="users"
)
secondary 指定中间表，多对多就是通过中间表关联两个表。
预加载关联数据：
from sqlalchemy.orm import selectinload, joinedload
stmt = select(User).options(selectinload(User.posts))
users = await db.execute(stmt)
stmt = select(Post).options(joinedload(Post.author))
posts = await db.execute(stmt)
selectinload 是额外查一次（两条 SQL），joinedload 是 JOIN 一次性查（一条 SQL）。一对多场景用 selectinload 更安全，不会出现数据膨胀。
过滤、排序、分页
from sqlalchemy import select, and_, or_, desc
stmt = select(User).where(User.age > 18)
stmt = select(User).where(and_(User.age > 18, User.is_active == True))
stmt = select(User).where(or_(User.role == "admin", User.role == "superuser"))
stmt = select(User).where(User.id.in_([1, 2, 3, 4]))
stmt = select(User).where(User.username.like("%admin%"))
stmt = select(User).order_by(desc(User.created_at))
page = 2
per_page = 20
stmt = select(User).offset((page-1)*per_page).limit(per_page)
聚合查询：
from sqlalchemy import func
total = await db.execute(select(func.count()).select_from(User))
stmt = select(User.is_active, func.count(User.id)).group_by(User.is_active)
results = await db.execute(stmt)
异步：2.0 最重要的特性
SQLAlchemy 2.0 的异步是原生的，不是靠线程池模拟的。
async def get_user_stats(db):
stmt = select(User.is_active, func.count(User.id)).group_by(User.is_active)
result = await db.execute(stmt)
return result.all()
async def batch_update(db, user_ids: list, is_active: bool):
stmt = select(User).where(User.id.in_(user_ids))
result = await db.execute(stmt)
users = result.scalars().all()
for user in users:
user.is_active = is_active
await db.commit()
FastAPI 集成：
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
app = FastAPI()
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
user = await get_user_by_id(db, user_id)
if not user:
return {"error": "not found"}
return user
@app.post("/users")
async def create_user(username: str, db: AsyncSession = Depends(get_db)):
return await create_user(db, username)
生产环境配置
连接池配置：
engine = create_async_engine(
"postgresql+asyncpg://user:pass@localhost/db",
pool_size=10,
max_overflow=20,
pool_timeout=30,
pool_recycle=3600,
pool_pre_ping=True,
)
事务管理：
async def transfer_money(db, from_id, to_id, amount):
try:
from_user = await db.execute(
select(User).where(User.id == from_id).with_for_update()
)
await db.commit()
except Exception:
await db.rollback()
raise
数据库迁移（Alembic）：
pip install alembic
alembic init alembic
alembic revision --autogenerate -m "add user table"
alembic upgrade head
性能优化：N+1 和批量操作
N+1 查询：最常见的问题
users = await db.execute(select(User))
for user in users.scalars():
print(len(user.posts))
stmt = select(User).options(selectinload(User.posts))
users = await db.execute(stmt)
批量插入：比逐条插入快几十倍
from sqlalchemy import insert
for i in range(10000):
db.add(User(username=f"user_{i}"))
await db.commit()
stmt = insert(User).values([{"username": f"user_{i}"} for i in range(10000)])
await db.execute(stmt)
await db.commit()
只查需要的字段，别 SELECT *：
stmt = select(User.id, User.username).where(User.is_active == True)
