跳到主要内容

SQLAlchemy

问题​

SQLAlchemy 的 Core 和 ORM 有什么区别？Session 的生命周期是怎样的？

答案​

SQLAlchemy 分为两层：Core（SQL 表达式语言）和 ORM（对象关系映射）。

Core vs ORM​

from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, select

engine = create_engine("postgresql+asyncpg://user:pass@localhost/db")
metadata = MetaData()

# Core：直接操作 SQL 表达式
users = Table("users", metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(50)),
)

# Core 查询
stmt = select(users).where(users.c.name == "Alice")
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

# ORM：定义模型类
class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str | None]

# ORM 查询（2.0 风格）
with Session(engine) as session:
    user = session.execute(
        select(User).where(User.name == "Alice")
    ).scalar_one_or_none()

关系映射​

from sqlalchemy.orm import relationship
from sqlalchemy import ForeignKey

class Department(Base):
    __tablename__ = "departments"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]

    # 一对多关系
    employees: Mapped[list["Employee"]] = relationship(back_populates="department")

class Employee(Base):
    __tablename__ = "employees"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    department_id: Mapped[int] = mapped_column(ForeignKey("departments.id"))

    department: Mapped["Department"] = relationship(back_populates="employees")

Session 生命周期​

Session 不是线程安全的

每个线程/请求应使用独立的 Session。Web 应用中通常用 scoped_session 或依赖注入管理。

异步支持（2.0+）​

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
async_session = async_sessionmaker(engine, class_=AsyncSession)

async def get_user(user_id: int) -> User | None:
    async with async_session() as session:
        result = await session.execute(
            select(User).where(User.id == user_id)
        )
        return result.scalar_one_or_none()

常见面试问题​

Q1: SQLAlchemy 1.x 和 2.0 的主要区别？​

答案：

特性1.x2.0
查询语法session.query(User)select(User)
模型定义Column(Integer)Mapped[int] = mapped_column()
类型提示弱完整的 Mapped 泛型
异步实验性正式支持

Q2: lazy loading vs eager loading？​

答案：

# lazy loading（默认）：访问关系属性时才查询，可能导致 N+1
user.posts  # 触发额外 SQL

# eager loading：一次查询加载关联数据
from sqlalchemy.orm import joinedload, selectinload

# JOIN 加载
stmt = select(User).options(joinedload(User.posts))

# 子查询加载（推荐多对多）
stmt = select(User).options(selectinload(User.posts))

Q3: 如何避免 N+1 查询问题？​

答案：

  1. 使用 selectinload / joinedload 预加载关联
  2. 设置 lazy="selectin" 默认策略
  3. 使用 sa.event.listen 监听 SQL 数量，开发环境检测 N+1
  4. 使用 raiseload 禁止隐式懒加载，强制显式声明

Q4: 如何处理事务？​

答案：

async with async_session() as session:
    async with session.begin():  # 自动 commit 或 rollback
        user = User(name="Alice")
        session.add(user)
    # begin 块结束时自动 commit
    # 如果抛异常，自动 rollback

Q5: flush 和 commit 的区别？​

答案：

  • flush()：将内存中的变更同步到数据库（执行 SQL），但不提交事务。其他连接看不到变更
  • commit()：先 flush()，再提交事务。变更对其他连接可见

相关链接​