上周同事发我一个导出脚本，说 2000 条订单跑了 40 秒。打开一看，主循环里每个订单查一次用户表。改成一次 IN 查询，总耗时掉到 1 秒以内。
这类问题不是 SQL 写得烂，是不知道 ORM 什么时候偷偷发了 SQL、发了几条。下面从零写一个能跑的最小 ORM，把 SQLAlchemy 的三块核心——描述符映射、Session 的 identity map、flush 的时机——拆开看，最后说清楚它在哪里会失效。
先看那段崩掉的代码长什么样：
# 反例：2000 条订单，2001 条 SQL
cur = conn.execute('SELECT id, user_id, amount FROM orders')
for oid, uid, amount in cur.fetchall():
name = conn.execute('SELECT name FROM users WHERE id = ?', (uid,)).fetchone()[0]
print(oid, name, amount)
SQLite 本地一条点查 20~30 微秒，2001 次也就 50 毫秒，你根本感觉不到。换到 MySQL，同机房一次往返 0.3ms，2000 次是 0.6 秒；跨机房 2ms，就是 4 秒。再加上连接池争用、结果集解析、Python 层的循环开销，几十秒不奇怪。
手写 SQL 要解决三件重复劳动：
- 行 → 对象。取出来的 tuple 按下标拆，加一列就全乱。
- 改对象 → 发 SQL。你改了
user.name，得自己记得写 UPDATE，还得只更新改过的字段。 - 同一行只有一个对象。两个地方各查一次同一行，改了一个忘了另一个，写回去就是丢更新。
ORM 就是这三件事的通用解法。SQLAlchemy 对应的是：描述符 + 元类、dirty tracking + unit of work、identity map。
映射层：为什么必须是描述符，不能是类属性
新手写 ORM 的第一版通常长这样：
class User:
id = None
name = None
这些是类属性，所有实例共享。u1.name = 'a' 会在实例的 __dict__ 里建一条新记录，看起来能用。但你没法在类上拿到「列对象」去拼 WHERE 子句，也没法在赋值时打脏标记。
正确做法是描述符：
class Column:
def __init__(self, type_, primary_key=False, nullable=True):
self.type = type_
self.primary_key = primary_key
self.nullable = nullable
self.name = None
def __set_name__(self, owner, name):
self.name = name
def __get__(self, obj, cls):
if obj is None:
return self # User.name 拿到 Column 对象本身
return obj._data.get(self.name)
def __set__(self, obj, value):
if obj._data.get(self.name) != value:
obj._dirty.add(self.name) # 赋值时打标记
obj._data[self.name] = value
__get__ 里 if obj is None 这个分支是整个设计的关键：类访问返回列对象，实例访问返回值。SQLAlchemy 的 InstrumentedAttribute 干的就是这件事，区别在于它返回的 QueryableAttribute 重载了 __eq__、__gt__、like 这些运算符，所以你能写 User.name == 'bob' 生成一个表达式对象。
这里有个很多人踩过的坑：为了让 user == 'bob' 能用，去重载实例的 __eq__。
# 错误写法
class User(Model):
def __eq__(self, other):
if isinstance(other, str):
return self.name == other
return super().__eq__(other)
Python 里定义了 __eq__ 会把 __hash__ 置成 None，对象直接不可 hash。后果是 identity map 不能用它当 key，塞进 set() 去重报 TypeError: unhashable type，更隐蔽的是两个字段完全相同的对象在集合里变成两条。
SQLAlchemy 从不在实例层面碰 __eq__，运算符重载全在描述符返回的对象上做，实例之间保持默认的身份相等。这是个很克制的决定，代价是你不能写 user == 'bob'，只能写 User.name == 'bob'。
元类负责把列收集起来：
from collections import OrderedDict
REGISTRY = []
class ModelMeta(type):
def __new__(mcls, name, bases, ns):
cls = super().__new__(mcls, name, bases, ns)
if not bases: # Model 基类本身跳过
return cls
cols = OrderedDict()
for base in reversed(cls.__mro__): # 支持继承，子类覆盖父类
for k, v in vars(base).items():
if isinstance(v, Column):
cols[v.name or k] = v
cls.__columns__ = cols
cls.__table__ = ns.get('__tablename__', name.lower())
cls.__pk__ = next(n for n, c in cols.items() if c.primary_key)
REGISTRY.append(cls)
return cls
Session：ORM 和「会发 SQL 的字典」的分界线
有了映射还不够。真正让 SQLAlchemy 区别于一堆辅助函数的是 Session。它内部维护三样东西：
class Session:
def __init__(self, conn):
self.conn = conn
self.identity_map = {} # (class, pk) -> instance
self._new = [] # 待 INSERT
self._deleted = [] # 待 DELETE
def get(self, cls, pk):
key = (cls, pk)
if key in self.identity_map:
return self.identity_map[key] # 命中，零 SQL
row = self.conn.execute(
f'SELECT * FROM {cls.__table__} WHERE {cls.__pk__} = ?', (pk,)
).fetchone()
if row is None:
return None
return self._from_row(cls, row)
def _from_row(self, cls, row):
data = dict(row) # 需要 conn.row_factory = sqlite3.Row
obj = cls(**data)
obj._persisted = True
self.identity_map[(cls, data[cls.__pk__])] = obj
return obj
identity map 带来三个白拿的好处：
s.get(User, 1) is s.get(User, 1)为 True。改一处，所有引用都看得见。- 同一行只发一次 SELECT。被十处引用也只有一次数据库往返。
- flush 时能精确算出哪些对象要写回，不用全表比对。
代价是内存。identity map 持有强引用，直到 session 关闭或者 expunge。批处理 100 万行还赖在同一个 session 里，内存一路涨到 OOM——这是后面要讲的坑。
flush 是另一块：
def flush(self):
for obj in self._new: # INSERT
self._insert(obj)
self._new.clear()
for obj in self._deleted: # DELETE
self.conn.execute(
f'DELETE FROM {type(obj).__table__} WHERE {type(obj).__pk__} = ?', (obj.pk,))
self.identity_map.pop((type(obj), obj.pk), None)
self._deleted.clear()
for (cls, pk), obj in list(self.identity_map.items()): # UPDATE
if not obj._dirty:
continue
sets = ', '.join(f'{n} = ?' for n in obj._dirty)
params = [obj._data[n] for n in obj._dirty] + [pk]
self.conn.execute(f'UPDATE {cls.__table__} SET {sets} WHERE {cls.__pk__} = ?', params)
obj._dirty.clear()
def commit(self):
self.flush()
self.conn.commit()
注意 flush 不发 commit。事务边界和 SQL 生成是两回事，这个分离是 SQLAlchemy 里最容易被误解的地方之一。
因为 INSERT 推迟到 flush 才发，下面这段会让人怀疑人生：
u = User(name='bob')
s.add(u)
print(s.query(User).filter('name', '=', 'bob').all()) # []，因为还没 INSERT
s.commit() # 这一刻才真正写库
SQLAlchemy 用 autoflush 兜这个坑：执行查询前先自动 flush 一次。默认开着。你可以 Session(autoflush=False) 关掉，在只读事务或者大批量导入时能省掉大量无谓的 flush，代价就是要自己记得手动 flush。
还有一处我的实现故意偷了懒：脏值判定用的是 __set__ 里打的布尔标记。SQLAlchemy 不是这么做的，它在赋值时把旧值记进 committed_state，flush 时拿当前值和快照比。区别在于 obj.name = obj.name 这种赋值——标记位会产生一条无意义的 UPDATE，快照比较不会。
flush 顺序：外键约束为什么突然炸了
我的最小实现按 add 顺序 INSERT。写个例子就露馅：
s.add(Order(user_id=1, amount=99)) # 先 add 子表
s.add(User(id=1, name='bob'))
s.commit()
# sqlite3.IntegrityError: FOREIGN KEY constraint failed
SQLAlchemy 不会炸，因为它维护了表之间的依赖图，flush 时按拓扑序插入，父表永远先。
自引用外键（无限级分类、树形结构里 parent_id 指向同表 id）拓扑排序无解，SQLAlchemy 会退化成按主键排序，或者要求你给关系加 post_update=True，把这条外键拆成第二条 UPDATE 发出去。
几个边界条件值得记一下：
- 环形依赖（A 引用 B，B 引用 A）在 SQLAlchemy 里直接抛
CircularDependencyError，只能靠post_update或者把其中一条设成nullable=True分两步走。 - 数据库层面的延迟约束（PostgreSQL 的
DEFERRABLE INITIALLY DEFERRED）能把检查推迟到 commit，但 SQLite 不支持，写了也白写。 - 我的最小实现没有拓扑排序能力，所以示例里必须父表先 add。这就是「最小可用」和「生产可用」的分界线。
N+1 到底是在哪一行产生的
换回真实的 SQLAlchemy，关系加载是另一个高频事故区。
from sqlalchemy import ForeignKey, Numeric, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = 'users'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(64), index=True)
orders: Mapped[list['Order']] = relationship(back_populates='user')
class Order(Base):
__tablename__ = 'orders'
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey('users.id'), index=True)
amount: Mapped[float] = mapped_column(Numeric(12, 2))
user: Mapped[User] = relationship(back_populates='orders')
# 1001 条 SQL：1 条查订单，1000 条查用户
orders = session.query(Order).limit(1000).all()
for o in orders:
print(o.user.name)
打开 create_engine(..., echo=True) 能直接看到这 1001 条。修复方式：
from sqlalchemy import select
from sqlalchemy.orm import selectinload
# 2 条 SQL：一条查订单，一条 WHERE users.id IN (...)
orders = session.execute(
select(Order).options(selectinload(Order.user)).limit(1000)
).scalars().all()
三种加载方式的取舍：
| 方式 | SQL 条数 | 适合 | 代价 |
|---|---|---|---|
| 默认 lazy | 1 + N | 只读单个对象的关联 | N+1 |
joinedload |
1 | 多对一、一对一 | 一对多会行爆炸，要做 DISTINCT 去重 |
selectinload |
2 | 一对多、多对多 | IN 列表太长会超参数上限（SQLite 旧版 999） |
raiseload |
1 | 生产环境兜底 | 需要显式预加载 |
joinedload 用在一对多上是个典型误用：1000 个订单平均每个 5 条明细，结果集变 5000 行，ORM 还得按主键把重复的父对象合并回去，内存和 CPU 双亏。一对多一律 selectinload。
想从根上防住，把忘记预加载从性能问题变成运行时异常：
class Order(Base):
...
user: Mapped[User] = relationship(lazy='raise') # 忘了 selectinload 就直接抛异常
在测试环境跑一遍，比上线后看慢查询日志强。
批量插入：ORM 最不该接的活
自己跑一下这份脚本，比较三种写法的量级差异：
import time, sqlite3
from sqlalchemy import create_engine, insert, Integer, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
class Base(DeclarativeBase):
pass
class Row(Base):
__tablename__ = 'rows'
id: Mapped[int] = mapped_column(Integer, primary_key=True)
a: Mapped[int] = mapped_column(Integer)
b: Mapped[str] = mapped_column(String(32))
N = 50_000
engine = create_engine('sqlite:///bench.db')
Base.metadata.create_all(engine)
# 1) 逐条 add
with Session(engine) as s:
t = time.perf_counter()
for i in range(N):
s.add(Row(a=i, b='x'))
s.commit()
print('add:', round(time.perf_counter() - t, 2), 's')
# 2) 字典列表直接 executemany
with Session(engine) as s:
t = time.perf_counter()
s.execute(insert(Row), [{'a': i, 'b': 'x'} for i in range(N)])
s.commit()
print('executemany:', round(time.perf_counter() - t, 2), 's')
我这边跑出来的量级差距在 20 倍上下，具体数字看磁盘。差距来源很清楚：第一种每个行都要构造实例、走描述符 __set__、塞进 identity map、flush 时逐条拼 INSERT；第二种直接交给 DBAPI 的 executemany。
| 方式 | 建实例 | 走 identity map | 主键回填 | 速度 |
|---|---|---|---|---|
session.add 逐条 |
是 | 是 | 是 | 最慢 |
bulk_insert_mappings |
否 | 否 | 部分 | 快 |
insert() + 字典列表 |
否 | 否 | 需 returning() |
最快 |
两个坑：
一，insert() 批量绕过了 ORM 事件和 identity map。 如果你之前 session.get(Row, 1) 过，内存里那个对象和库里新插的行就对不上，后续读到的还是旧值。批量和 ORM 混用时，记得 session.expire_all()。
二，别每 1000 行 commit 一次。 一次 commit 就是一次 fsync，机械盘上 1000 次够你喝杯咖啡。但事务也别开太大：PostgreSQL 长事务会阻挡 vacuum，undo 空间一路涨。10 万到 100 万行一批是个常见折中。
五个我踩过的坑
1. DetachedInstanceError
user = session.get(User, 1)
session.close()
print(user.orders) # DetachedInstanceError
懒加载需要一个活着的 session 来发 SQL。expire_on_commit=False 只解决属性过期，解决不了这个。要么在 session 内用 selectinload 加载完，要么别关 session。
2. Session 做成模块级单例
# 错误
session = Session(engine)
def get_user(uid):
return session.get(User, uid)
两个请求同时进来，identity map 互相污染，一个 rollback 影响另一个，事务边界完全乱套。正确姿势是每请求一个 session：
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)
def handle_request():
session = SessionLocal()
try:
do_work(session)
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
3. expire_on_commit=True 在脚本里变成 N+1
默认 commit 之后所有属性过期，下次访问重新 SELECT。Web 请求里一次 commit 无所谓，循环里就是灾难：
for o in session.query(Order).all():
session.commit()
print(o.amount) # 每次 commit 后再访问，又发一条 SELECT
批处理脚本里设 expire_on_commit=False。
4. 金额用 Float
amount = Column(Float) # 错误
amount = Column(Numeric(12, 2)) # 正确
SQLite 没有原生 decimal，Numeric 存进去的类型亲和性是 NUMERIC，值可能落成 INTEGER 或 REAL；PostgreSQL 是真 decimal。跨库迁移的账目核对能让你加班。稳妥点直接存整数分。
5. 表名撞 SQL 关键字
我的最小实现直接拼表名。class Order 对应的表叫 order，SELECT * FROM order 在 SQLite 里直接语法错误，因为 ORDER 是关键字。SQLAlchemy 的方言层用 quoted_name 和 identifier preparer 处理：MySQL 用反引号，PostgreSQL 用双引号。自己拼 SQL 就老老实实给标识符加引号。
完整代码
miniorm.py，纯标准库，可以直接跑：
# miniorm.py —— 用于理解原理的最小 ORM，不要上生产
import sqlite3
from collections import OrderedDict
REGISTRY = []
class Column:
def __init__(self, type_, primary_key=False, nullable=True):
self.type = type_
self.primary_key = primary_key
self.nullable = nullable
self.name = None
def __set_name__(self, owner, name):
self.name = name
def __get__(self, obj, cls):
if obj is None:
return self
return obj._data.get(self.name)
def __set__(self, obj, value):
if obj._data.get(self.name) != value:
obj._dirty.add(self.name)
obj._data[self.name] = value
class ModelMeta(type):
def __new__(mcls, name, bases, ns):
cls = super().__new__(mcls, name, bases, ns)
if not bases:
return cls
cols = OrderedDict()
for base in reversed(cls.__mro__):
for k, v in vars(base).items():
if isinstance(v, Column):
cols[v.name or k] = v
cls.__columns__ = cols
cls.__table__ = ns.get('__tablename__', name.lower())
cls.__pk__ = next(n for n, c in cols.items() if c.primary_key)
REGISTRY.append(cls)
return cls
class Model(metaclass=ModelMeta):
__columns__ = OrderedDict()
def __init__(self, **kw):
self._data = {}
self._dirty = set()
self._persisted = False
for k, v in kw.items():
setattr(self, k, v)
self._dirty.clear() # 构造赋值不算修改
@property
def pk(self):
return self._data.get(type(self).__pk__)
def __repr__(self):
return f'<{type(self).__name__} {self._data}>'
def create_all(conn):
for cls in REGISTRY:
cols = []
for name, c in cls.__columns__.items():
ddl = f'{name} {c.type}'
if c.primary_key:
ddl += ' PRIMARY KEY'
if c.type.upper() == 'INTEGER':
ddl += ' AUTOINCREMENT'
elif not c.nullable:
ddl += ' NOT NULL'
cols.append(ddl)
conn.execute(f'CREATE TABLE IF NOT EXISTS {cls.__table__} ({", ".join(cols)})')
conn.commit()
class Query:
def __init__(self, session, cls):
self.s = session
self.cls = cls
self._where, self._params = [], []
def filter(self, column, op, value):
if op not in ('=', '>', '<', '>=', '<=', '!=', 'LIKE'):
raise ValueError(op)
self._where.append(f'{column} {op} ?')
self._params.append(value)
return self
def all(self):
sql = f'SELECT * FROM {self.cls.__table__}'
if self._where:
sql += ' WHERE ' + ' AND '.join(self._where)
rows = self.s.conn.execute(sql, self._params).fetchall()
out = []
for r in rows:
data = dict(r)
key = (self.cls, data[self.cls.__pk__])
# 命中 identity map 就复用，不建新对象
out.append(self.s.identity_map.get(key) or self.s._from_row(self.cls, r))
return out
def first(self):
rows = self.all()
return rows[0] if rows else None
class Session:
def __init__(self, conn):
self.conn = conn
self.identity_map = {}
self._new = []
self._deleted = []
conn.execute('PRAGMA foreign_keys = ON')
def add(self, obj):
self._new.append(obj)
return obj
def delete(self, obj):
self._deleted.append(obj)
def query(self, cls):
return Query(self, cls)
def get(self, cls, pk):
key = (cls, pk)
if key in self.identity_map:
return self.identity_map[key]
row = self.conn.execute(
f'SELECT * FROM {cls.__table__} WHERE {cls.__pk__} = ?', (pk,)
).fetchone()
return None if row is None else self._from_row(cls, row)
def _from_row(self, cls, row):
data = dict(row)
obj = cls(**data)
obj._persisted = True
self.identity_map[(cls, data[cls.__pk__])] = obj
return obj
def _insert(self, obj):
cls = type(obj)
names, params = [], []
for name, col in cls.__columns__.items():
if col.primary_key and obj._data.get(name) is None:
continue # 自增主键不显式写入
names.append(name)
params.append(obj._data.get(name))
sql = (f'INSERT INTO {cls.__table__} ({", ".join(names)}) '
f'VALUES ({", ".join(["?"] * len(names))})')
cur = self.conn.execute(sql, params)
if obj._data.get(cls.__pk__) is None:
obj._data[cls.__pk__] = cur.lastrowid
obj._persisted = True
self.identity_map[(cls, obj.pk)] = obj
def flush(self):
for obj in self._new:
self._insert(obj)
self._new.clear()
for obj in self._deleted:
self.conn.execute(
f'DELETE FROM {type(obj).__table__} WHERE {type(obj).__pk__} = ?', (obj.pk,))
self.identity_map.pop((type(obj), obj.pk), None)
self._deleted.clear()
for (cls, pk), obj in list(self.identity_map.items()):
if not obj._dirty:
continue
sets = ', '.join(f'{n} = ?' for n in obj._dirty)
params = [obj._data[n] for n in obj._dirty] + [pk]
self.conn.execute(
f'UPDATE {cls.__table__} SET {sets} WHERE {cls.__pk__} = ?', params)
obj._dirty.clear()
def commit(self):
try:
self.flush()
self.conn.commit()
except Exception:
self.conn.rollback()
self._new.clear()
self.identity_map.clear()
raise
def close(self):
self.identity_map.clear()
self.conn.close()
跑一下，验证 identity map 和延迟 flush：
import sqlite3
from miniorm import Column, Model, Session, create_all
class User(Model):
__tablename__ = 'users'
id = Column('INTEGER', primary_key=True)
name = Column('TEXT', nullable=False)
class Order(Model):
__tablename__ = 'orders'
id = Column('INTEGER', primary_key=True)
user_id = Column('INTEGER', nullable=False)
amount = Column('INTEGER', nullable=False)
conn = sqlite3.connect(':memory:')
conn.row_factory = sqlite3.Row
create_all(conn)
s = Session(conn)
s.add(User(name='alice'))
s.add(Order(user_id=1, amount=9900))
s.commit()
u1 = s.get(User, 1)
u2 = s.get(User, 1)
print(u1 is u2) # True，identity map 命中
u1.name = 'bob'
u1.name = 'bob' # 重复赋值，_dirty 不会重复计数
s.flush() # 这一条 UPDATE 只在 flush 时发出
print(conn.execute('SELECT name FROM users WHERE id = 1').fetchone()[0]) # bob
__set__ 里那句 if obj._data.get(self.name) != value 顺带解决了重复赋值产生多余 UPDATE 的问题，不过它只挡住了「值没变」的情况，挡不住「改成 A 又改回原值」——那仍然会发一条 UPDATE。生产实现里这一层会用快照比对，代价是每次 flush 要遍历所有属性。
这套东西的价值不在于「我也能写 ORM」。下次查询日志里冒出 1001 条 SQL、看到 DetachedInstanceError、或者发现 10 万行导入跑了五分钟，你能立刻对上是哪个设计决策出了问题，以及该在哪一层修——是补 selectinload，是换 expire_on_commit，还是干脆绕开 ORM 走 executemany。