SQLAlchemy 2.1 上手：Core 与 ORM 的最小组合，加四条真实报错

同一个项目里，报表类查询想直接写 SQL 拿行，业务逻辑想用对象和关系。SQLAlchemy 把这两层都给了：Core 用 select()、insert() 这些构造器拼语句，返回的是行；ORM 用声明式模型和 Session 管理对象状态、事务、关系加载。代价是概念多，2.0 之后又把 session.query() 换成了 select() + session.scalars()，老代码和新文档混着看的阶段最容易出错。如果只有一张表、几个字段、几条 SQL，sqlite3 模块加参数化查询更直接。

安装与实际版本

python -m venv D:\ai\cache\temp\py-sqlalchemy
D:\ai\cache\temp\py-sqlalchemy\Scripts\python.exe -m pip install sqlalchemy

Name: SQLAlchemy
Version: 2.1.8
Location: D:\ai\cache\temp\py-sqlalchemy\Lib\site-packages

import sqlalchemy as sa

print(sa.__version__)

2.1.8

示例全部落在本地 SQLite 文件上，不连外部数据库。

Core：建表、插入、查询

import os
import tempfile

import sqlalchemy as sa
from sqlalchemy import String, create_engine, select

db = os.path.join(tempfile.gettempdir(), "kernelranch_sa_demo.db")
engine = create_engine(f"sqlite:///{db}")
meta = sa.MetaData()
books = sa.Table(
    "books",
    meta,
    sa.Column("id", sa.Integer, primary_key=True),
    sa.Column("title", String(64), nullable=False),
    sa.Column("price", sa.Numeric(8, 2), nullable=False),
    sa.Column("stock", sa.Integer, default=0),
)
meta.create_all(engine)

with engine.begin() as conn:
    conn.execute(
        books.insert(),
        [
            {"title": "数据库随笔", "price": 59.00, "stock": 12},
            {"title": "HTTP 手记", "price": 39.50, "stock": 0},
            {"title": "终端美学", "price": 45.00, "stock": 7},
        ],
    )
    rows = conn.execute(select(books.c.title, books.c.price).order_by(books.c.price)).all()
    print("Core 查询:", rows)
    print("row 可当元组:", rows[0][0], "| 也可按列名:", rows[0].title)

Core 查询: [('HTTP 手记', Decimal('39.50')), ('终端美学', Decimal('45.00')), ('数据库随笔', Decimal('59.00'))]
row 可当元组: HTTP 手记 | 也可按列名: HTTP 手记

Numeric 拿回来是 Decimal 而不是 float，金额场景这点很重要。engine.begin() 是自带事务的上下文，正常退出提交、异常回滚。

构造器可以编译成 SQL 看，排查"ORM 到底发了什么"时很有用：

stmt = (
    select(books.c.title, books.c.price)
    .where(books.c.stock > 0)
    .order_by(books.c.price.desc())
    .limit(2)
)
print(str(stmt.compile(engine, compile_kwargs={"literal_binds": True})))

SELECT books.title, books.price 
FROM books 
WHERE books.stock > 0 ORDER BY books.price DESC
 LIMIT 2 OFFSET 0

ORM：声明模型与关系

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


class Base(DeclarativeBase):
    pass


class Author(Base):
    __tablename__ = "authors"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(32))
    books: Mapped[list["Book"]] = relationship(back_populates="author", cascade="all, delete-orphan")


class Book(Base):
    __tablename__ = "orm_books"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(64))
    price: Mapped[float]
    author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))
    author: Mapped[Author] = relationship(back_populates="books")


Base.metadata.create_all(engine)

with Session(engine) as s:
    a = Author(name="林可")
    a.books = [Book(title="数据库随笔", price=59.0), Book(title="HTTP 手记", price=39.5)]
    s.add(a)
    s.commit()
    print("插入后 author.id:", a.id, "首本书 id:", a.books[0].id)

    one = s.scalars(select(Author).where(Author.name == "林可")).one()
    print("one():", one.name, "| 书:", [b.title for b in one.books])
    total = s.scalar(select(sa.func.count()).select_from(Book))
    print("scalar(count):", total)

插入后 author.id: 1 首本书 id: 1
one(): 林可 | 书: ['数据库随笔', 'HTTP 手记']
scalar(count): 2

关系赋值时 Session 会自动按外键顺序插入，父表拿到自增 id 后再写子表，不需要手动 flush。select(Author) 返回的是实体，取单值用 scalars() 而不是 execute()，后者拿到的是 Row。

常用 API 与参数

四个坑

第一个坑是外键写错位置。把 ForeignKey 直接赋给 Mapped 注解，映射配置阶段就会失败：

sqlalchemy.exc.NoForeignKeysError: Could not determine join condition between parent/child tables on relationship Author.books - there are no foreign keys linking these tables.  Ensure that referencing columns are associated with a ForeignKey or ForeignKeyConstraint, or specify a 'primaryjoin' expression.

报错说的是"没有外键"，但代码里明明写了 ForeignKey("authors.id")。原因是 Mapped[int] = ForeignKey(...) 没有走 mapped_column()，这个 ForeignKey 对象没有绑定到列上。改成 mapped_column(ForeignKey("authors.id")) 就正常了。

第二个坑是会话关掉之后访问关系。懒加载需要活动的 Session：

Parent instance <Author at 0x2622e433990> is not bound to a Session; lazy load operation of attribute 'books' cannot proceed (Background on this error at: https://sqlalche.me/e/21/bhk3)

把对象带出 with 块、再在模板或序列化里访问 author.books，就会撞上这条。要么在会话内 selectinload 提前取好，要么把 expire_on_commit=False 配上并在关闭前把要用的属性都读出来。

第三个坑是懒加载的 N+1。5 个作者、每人 3 本书，遍历取 len(a.books)，用事件钩子数一下实际发出的 SQL：

懒加载：5 个作者、每个 3 本书
SQL 条数（懒加载）: 6
SQL 条数（selectinload）: 2

6 条是 1 条作者查询加 5 条按作者取书的查询；selectinload 把它压到 2 条。作者数从 5 涨到 500 时，前者是 501 条，后者还是 2 条。

第四个坑是类型映射的精度。Mapped[float] 建出来的列是 DOUBLE：

CREATE TABLE orm_books (
	id INTEGER NOT NULL, 
	title VARCHAR(64) NOT NULL, 
	price DOUBLE NOT NULL, 
	author_id INTEGER NOT NULL, 
	PRIMARY KEY (id), 
	FOREIGN KEY(author_id) REFERENCES authors (id)
)

金额用 DOUBLE 存会有二进制浮点的舍入问题。Core 那边写 sa.Numeric(8, 2) 拿回来是 Decimal，ORM 这边就要写 Mapped[Decimal] = mapped_column(sa.Numeric(8, 2))，别图省事用 float。

顺带一个 2.0 之后必踩的：直接 conn.execute("select 1") 不再接受字符串。

ObjectNotExecutableError: Not an executable object: 'select 1'
用 text() 才正确: 1

裸 SQL 要包一层 text()，这样参数绑定和方言处理才有地方挂。

小结

Core 与 ORM 不是二选一。报表、批量写入、需要精确控制 SQL 的地方用 Core，业务对象和关系用 ORM，两边共用一份 MetaData 就行。判断标准是"这段代码关心的是行还是对象"。

四个坑的共同点是报错位置和原因隔了一层：NoForeignKeysError 指向关系配置，根因在列定义；DetachedInstanceError 指向属性访问，根因在会话生命周期；N+1 不报错，只在 SQL 计数上体现；Mapped[float] 也不报错，只在存了钱之后体现。把 echo=True 和 SQL 计数打开，后面两个能提前看见。

数据来源与检索时间

检索时间 2026-11-10。版本、DDL、SQL 条数与报错文本均为本机实测。

核验说明

• 虚拟环境 D:\ai\cache\temp\py-sqlalchemy，Python 3.11.9，pip install sqlalchemy 装到 2.1.8。
• 数据库用本地 SQLite 文件（kernelranch_sa_demo.db、kernelranch_np1.db），没有连接任何外部数据库，示例里也没有方言相关的写法。
• 真跑的示例：Core 的建表/插入/查询与 compile() 输出、ORM 的插入、scalars().one()、scalar(count)、更新与删除、会话关闭后访问关系、裸字符串 SQL、SQL 计数对照。
• 没测的部分：Alembic 迁移、异步引擎（create_async_engine）、连接池参数调优、PostgreSQL 方言相关特性（RETURNING、JSONB、数组类型）、Session 的 expire_on_commit 对照实验。这些都没有在本机验证，文章里没给结论。
• N+1 的 SQL 条数用 before_cursor_execute 事件统计，只统计了这一条路径上的语句，不含 SQLite 自身的 PRAGMA。