SQLAlchemy

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. 查询统一了:1.x 里 Core 和 ORM 两套 API,2.0 统一用 select()
  2. 异步原生支持async/await 操作数据库,配合 FastAPI 很顺畅
  3. 类型提示完善Mapped + mapped_column,IDE 能帮你自动补全

准备工作:安装和连接

pip install sqlalchemy
pip install asyncpg  # PostgreSQL 异步驱动
pip install aiosqlite # SQLite 异步驱动(开发用)

同步连接(简单场景):

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)  # 刷新获取自增 id
    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"   # 预加载,避免 N+1
    )

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

# User 里加:
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"))

# IN 查询
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 的异步是原生的,不是靠线程池模拟的。

# 所有数据库操作都加 await
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_size 最多额外创建多少
    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 查询:最常见的问题

# 问题代码:先查用户列表(1次查询),每个用户访问 posts 时再查一次(N次)
users = await db.execute(select(User))
for user in users.scalars():
    print(len(user.posts))   # 触发 N 次额外查询

# 解决方案:用 selectinload 预加载
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 *

# 只需要 id 和 username,别查所有字段
stmt = select(User.id, User.username).where(User.is_active == True)