从纯文本到 SQLAlchemy：Python 数据管理的完整演进路径

预计阅读时间：10 分钟

很多 Python 开发者第一次接触"数据持久化"，是从一个 CSV 文件或 JSON 字典开始的。但随着业务变复杂——需要关联查询、事务保证、多表联动——纯文件方案很快就会撞墙。SQLite 和 SQLAlchemy 正是这条演进路上的两个关键节点：前者给你一个零配置的真实数据库，后者让你用 Python 对象的思维操作它。

这篇文章沿着"flat file → raw SQL → ORM"这条线，把每个阶段的核心问题和实操要点拆开，最后给出一套可以直接跑的 SQLAlchemy 多对多关系示例。

纯文件方案的天花板

用 CSV 或 JSON 存数据，起步确实快：

import csv

with open("users.csv", newline="") as f:
    rows = list(csv.DictReader(f))

读出来是字典列表，写回去也简单。但问题很快出现：

并发写入冲突——两个进程同时写，数据互相覆盖。
关联查询困难——想查"某个用户的所有订单"，要手动遍历两张列表做匹配，O(n²) 复杂度。
无事务保证——写入中途崩溃，文件可能处于半损坏状态。

当数据量超过几千行、或者需要多表关联时，就该上 SQL 了。

SQLite：零配置的真实数据库

Python 标准库自带 sqlite3 模块，不需要安装任何服务端：

import sqlite3

conn = sqlite3.connect("app.db")
cursor = conn.cursor()

cursor.execute("""
    CREATE TABLE IF NOT EXISTS user (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE NOT NULL
    )
""")
cursor.execute("""
    CREATE TABLE IF NOT EXISTS order (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        user_id INTEGER NOT NULL,
        amount REAL NOT NULL,
        FOREIGN KEY (user_id) REFERENCES user(id)
    )
""")
conn.commit()

关键点：

AUTOINCREMENT 保证主键自增，不依赖手动赋值。
FOREIGN KEY 建立外键约束，SQLite 默认不启用外键检查，需要额外执行 PRAGMA foreign_keys = ON。
UNIQUE 约束防止邮箱重复插入，违反时抛 IntegrityError。

插入和查询：

cursor.execute("INSERT INTO user (name, email) VALUES (?, ?)", ("张三", "zhang@example.com"))
cursor.execute("SELECT u.name, o.amount FROM user u JOIN order o ON u.id = o.user_id WHERE u.name = ?", ("张三",))
rows = cursor.fetchall()
conn.commit()

用 ? 占位符而非字符串拼接，是防止 SQL 注入的基本做法。

到这一步，你已经有了事务、约束、索引、关联查询——但代码里全是手写 SQL 字串，维护和重构成本开始上升。

SQLAlchemy Core：用 Python 表达式写 SQL

SQLAlchemy 分两层：Core（表达式层）和 ORM（对象层）。Core 层让你用 Python 对象描述表结构，再编译成 SQL：

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, Float, ForeignKey

engine = create_engine("sqlite:///app.db")
metadata = MetaData()

user_table = Table("user", metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("name", String, nullable=False),
    Column("email", String, unique=True, nullable=False),
)

order_table = Table("order", metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("user_id", Integer, ForeignKey("user.id"), nullable=False),
    Column("amount", Float, nullable=False),
)

metadata.create_all(engine)

查询时不再拼字符串：

from sqlalchemy import select

with engine.connect() as conn:
    stmt = select(user_table.c.name, order_table.c.amount).join_from(
        user_table, order_table, user_table.c.id == order_table.c.user_id
    ).where(user_table.c.name == "张三")
    result = conn.execute(stmt)
    for row in result:
        print(row)

Core 层的优势：SQL 语句由 Python 表达式组合而成，IDE 可以做类型检查和自动补全，重构时改列名不会漏掉某处硬编码的字符串。

SQLAlchemy ORM：用对象思维操作数据

ORM 层把表映射成 Python 类，行映射成实例：

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

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(nullable=False)
    email: Mapped[str] = mapped_column(unique=True, nullable=False)
    orders: Mapped[list["Order"]] = relationship(back_populates="user")

class Order(Base):
    __tablename__ = "order"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("user.id"), nullable=False)
    amount: Mapped[float] = mapped_column(nullable=False)
    user: Mapped["User"] = relationship(back_populates="orders")

Mapped + mapped_column 是 SQLAlchemy 2.0 的声明式写法，类型标注直接参与列定义，IDE 提示更友好。

操作数据：

from sqlalchemy import create_engine

engine = create_engine("sqlite:///app.db")
Base.metadata.create_all(engine)

with Session(engine) as session:
    u = User(name="张三", email="zhang@example.com")
    o1 = Order(amount=99.5, user=u)
    o2 = Order(amount=200.0, user=u)
    session.add_all([u, o1, o2])
    session.commit()

    # 查询：通过关系直接访问
    zhang = session.execute(select(User).where(User.name == "张三")).scalar_one()
    for o in zhang.orders:
        print(f"订单 {o.id}: ¥{o.amount}")

注意 relationship 的 back_populates 参数——它让双向导航生效：从 User 能拿到 .orders，从 Order 能拿到 .user。漏写这个参数，反向访问会返回空列表或 None。

多对多关系：中间表 + relationship

这是数据管理知识里最容易卡住的一环。假设"用户"和"标签"是多对多：

from sqlalchemy import Table, Column, Integer, ForeignKey, String, create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session
from sqlalchemy import select

class Base(DeclarativeBase):
    pass

# 中间表：纯 Table 对象，不需要映射成类
user_tag = Table(
    "user_tag", Base.metadata,
    Column("user_id", Integer, ForeignKey("user.id"), primary_key=True),
    Column("tag_id", Integer, ForeignKey("tag.id"), primary_key=True),
)

class User(Base):
    __tablename__ = "user"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    tags: Mapped[list["Tag"]] = relationship(
        secondary=user_tag, back_populates="users", lazy="selectin"
    )

class Tag(Base):
    __tablename__ = "tag"
    id: Mapped[int] = mapped_column(primary_key=True)
    label: Mapped[str] = mapped_column(String(30), unique=True)
    users: Mapped[list["User"]] = relationship(
        secondary=user_tag, back_populates="tags", lazy="selectin"
    )

engine = create_engine("sqlite:///m2m.db", echo=True)
Base.metadata.create_all(engine)

with Session(engine) as session:
    t1 = Tag(label="Python")
    t2 = Tag(label="数据库")
    u1 = User(name="张三", tags=[t1, t2])
    u2 = User(name="李四", tags=[t1])
    session.add_all([u1, u2, t1, t2])
    session.commit()

    # 反查：某个标签下的所有用户
    python_tag = session.execute(select(Tag).where(Tag.label == "Python")).scalar_one()
    print(f"标签 '{python_tag.label}' 下的用户:", [u.name for u in python_tag.users])

几个实操要点：

中间表用 Table 而非 ORM 类——它只存两个外键，没有业务字段，不需要独立生命周期。
secondary 参数指向中间表对象，relationship 自动处理 JOIN。
lazy="selectin"——默认的 lazy="select" 会在访问关系属性时再发一条查询（N+1 问题）。selectin 用 IN 子查询一次性加载关联数据，对多对多场景更合理。
echo=True——开发阶段打开，终端会打印所有生成的 SQL，方便确认 ORM 行为是否符合预期。生产环境务必关掉。

选型建议与常见陷阱

场景	推荐方案
几百行数据、单进程读写、无关联查询	CSV / JSON 足够
需要事务、约束、索引，但不想部署数据库服务	SQLite + `sqlite3` 模块
多表关联频繁，想减少手写 SQL 维护成本	SQLAlchemy Core
业务对象有复杂生命周期，希望用 Python 类封装	SQLAlchemy ORM

常见陷阱清单：

SQLite 外键默认不生效——每次连接后执行 PRAGMA foreign_keys = ON，或在 create_engine 里加事件监听自动执行。
ORM 的 N+1 查询——循环里访问 user.orders，每次都发一条 SELECT。用 selectin 或 joinedload 预加载。
Session 未关闭——Session 是有生命周期的对象，用 with Session(engine) as session: 确保退出时自动关闭。
多进程写 SQLite——SQLite 写锁是文件级的，多进程并发写会频繁超时。要么改用 PostgreSQL，要么把写操作收敛到单进程。

从纯文件到 ORM，每一步都是对前一步瓶颈的回应。理解这条演进路径，比记住某个 API 的参数更重要——因为下一个项目的数据规模，决定你该停在哪个阶段。