SQLAlchemy Core 与 Engine

第 9 章 · SQLAlchemy 2.0 Core 与 Engine

本章目标：掌握 SQLAlchemy 2.0 Core（非 ORM）层：Engine、Connection、text() 与 insert()/update() 构造器；理解 显式事务 与 begin() 语义；在 db-demo 中用 Core 重写 ch08 部分 CRUD；保持 slug + is_published + price（分） 字段约定；为 ch10 ORM 与 ch11 Alembic 铺路。

学时建议：5～6 小时（含 2 小时 Core 跟练）

前置：完成 python-database ch07～ch08；了解 SQL 与 Python 类型注解。

9.1 场景说明：Core 介于 raw SQL 与 ORM 之间

Core 适合：复杂 SQL、跨数据库迁移、不建 Model 的临时任务。

db-demo 数据访问栈
  ch08 sqlite3 ──► ch09 Core ──► ch10 ORM ──► ch11 Alembic

9.2 安装与项目布局

cd ~/python-learn/db-demo
python -m venv .venv
source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install "sqlalchemy>=2.0,<3"
pip freeze | grep -i sqlalchemy >> requirements.txt

db-demo/
├── core/
│   ├── __init__.py
│   ├── engine.py          # Engine 工厂
│   ├── schema.py          # Table 元数据（Core 反射或声明）
│   └── product_ops.py     # insert/update 示例
├── data/shop_db.sqlite3
└── requirements.txt

连接 URL：

# SQLite 同步（本章）
DATABASE_URL = "sqlite:///data/shop_db.sqlite3"

# 生产 MySQL 示例（仅注释）
# mysql+pymysql://user:pass@localhost/shop_db?charset=utf8mb4

9.3 Engine：连接池入口

core/engine.py：

from pathlib import Path
from sqlalchemy import create_engine, event
from sqlalchemy.engine import Engine

ROOT = Path(__file__).resolve().parents[1]
DB_FILE = ROOT / "data" / "shop_db.sqlite3"
DATABASE_URL = f"sqlite:///{DB_FILE.as_posix()}"


def make_engine(echo: bool = False) -> Engine:
    eng = create_engine(
        DATABASE_URL,
        echo=echo,                 # True 打印 SQL
        future=True,               # 2.0 风格（2.0 默认 True）
        pool_pre_ping=True,
    )

    @event.listens_for(eng, "connect")
    def _set_sqlite_pragma(dbapi_conn, _connection_record):
        cursor = dbapi_conn.cursor()
        cursor.execute("PRAGMA foreign_keys = ON")
        cursor.close()

    return eng


engine = make_engine()

Engine vs Connection：

9.4 Connection 与 text() 查询

from sqlalchemy import text
from core.engine import engine

with engine.connect() as conn:
    result = conn.execute(
        text("""
            SELECT p.slug, p.title, p.price, p.is_published
            FROM products p
            INNER JOIN categories c ON c.id = p.category_id
            WHERE c.slug = :category_slug AND p.is_published = 1
            ORDER BY p.price ASC
            LIMIT :limit OFFSET :offset
        """),
        {"category_slug": "digital", "limit": 10, "offset": 0},
    )
    rows = result.mappings().all()   # list[RowMapping]，类 dict
    for row in rows:
        print(row["slug"], row["price"])  # price 为分

connect() 默认不自动 commit；只读可不带事务；写入需 §9.7 事务。

9.5 Table 元数据（Core 声明表）

与 ch07 DDL 对齐，用 Core 描述（供 insert/update 构造器使用）：

core/schema.py：

from sqlalchemy import (
    MetaData, Table, Column,
    Integer, Text, ForeignKey, CheckConstraint,
)

metadata = MetaData()

categories = Table(
    "categories", metadata,
    Column("id", Integer, primary_key=True),
    Column("name", Text, nullable=False),
    Column("slug", Text, nullable=False, unique=True),
    Column("sort_order", Integer, nullable=False, server_default="0"),
)

products = Table(
    "products", metadata,
    Column("id", Integer, primary_key=True),
    Column("category_id", Integer, ForeignKey("categories.id"), nullable=False),
    Column("title", Text, nullable=False),
    Column("slug", Text, nullable=False, unique=True),
    Column("description", Text),
    Column("price", Integer, nullable=False),       # 分
    Column("stock", Integer, nullable=False, server_default="0"),
    Column("is_published", Integer, nullable=False, server_default="0"),
    CheckConstraint("price >= 0", name="ck_products_price_nonneg"),
    CheckConstraint("is_published IN (0, 1)", name="ck_products_published"),
)

本章用 声明 Table 演示 Core DML，表已由 ch07 SQL 创建。

9.6 insert() 与 returning

SQLAlchemy 2.0 风格：

from sqlalchemy import insert, select
from core.engine import engine
from core.schema import products

def create_product(
    category_id: int,
    title: str,
    slug: str,
    price_cents: int,
    stock: int,
    is_published: bool = False,
) -> int:
    stmt = (
        insert(products)
        .values(
            category_id=category_id,
            title=title,
            slug=slug,
            price=price_cents,
            stock=stock,
            is_published=int(is_published),
        )
        .returning(products.c.id)
    )
    with engine.begin() as conn:          # 自动 commit
        new_id = conn.execute(stmt).scalar_one()
    return new_id

批量插入：

from sqlalchemy import insert

rows = [
    {"category_id": 1, "title": "A", "slug": "a", "price": 1000, "stock": 1, "is_published": 1},
    {"category_id": 1, "title": "B", "slug": "b", "price": 2000, "stock": 1, "is_published": 1},
]
with engine.begin() as conn:
    conn.execute(insert(products), rows)