使用attrs+SQLAlchemy命令式映射插入外键时触发非空约束错误

安全阿岳

2026-6-13

问题

我正在构建一个小型Python应用以学习领域驱动设计（DDD）方法，使用attrs类作为领域模型，并通过命令式方式构建ORM实例。但插入包含外键的类时遇到错误：

sqlalchemy.exc.IntegrityError: (sqlite3.IntegrityError) NOT NULL constraint failed: some_events.measurement_log_before_id
[SQL: INSERT INTO some_events (created_at, event, measurement_log_before_id, measurement_log_after_id) VALUES (?, ?, ?, ?)]
[parameters: ('2025-05-05 20:13:06.936736', 'some event', None, None)]

奇怪的是，attrs实例中存在id值，但ORM类在session.commit()过程中丢失了该值。

可复现代码

from datetime import UTC, datetime, timedelta
from pathlib import Path

import pytest
from attrs import define, field
from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, Table, func
from sqlalchemy.engine import create_engine
from sqlalchemy.orm import Session, registry, relationship, sessionmaker

mapper_registry = registry()
metadata = mapper_registry.metadata

measurement_table = Table(
    "measurement_log",
    metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("recorded_at", DateTime, nullable=False),
    Column("measurement", Integer, nullable=False),
)


some_event_table = Table(
    "some_events",
    metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("created_at", DateTime, nullable=False, server_default=func.now()),
    Column("event", String, nullable=False),
    Column(
        "measurement_log_before_id", ForeignKey("measurement_log.id"), nullable=False
    ),
    Column(
        "measurement_log_after_id", ForeignKey("measurement_log.id"), nullable=False
    ),
)


@define(order=True, slots=False)
class MeasurementLog:
    id: int = field(init=False)
    recorded_at: datetime
    measurement: int


@define(slots=False)
class SomeEvents:
    id: int = field(init=False)
    measurement_log_before_id: int
    measurement_log_after_id: int
    consumer: str
    event: str
    created_at: datetime = field(default=datetime.now(UTC))
    before_log: MeasurementLog | None = None
    after_log: MeasurementLog | None = None


mapper_registry.map_imperatively(
    MeasurementLog,
    measurement_table,
    properties={
        "starts_event": relationship(
            SomeEvents,
            foreign_keys=[some_event_table.c.measurement_log_before_id],
            back_populates="before_log",
        ),
        "ends_event": relationship(
            SomeEvents,
            foreign_keys=[some_event_table.c.measurement_log_after_id],
            back_populates="after_log",
        ),
    },
)

mapper_registry.map_imperatively(
    SomeEvents,
    some_event_table,
    properties={
        "before_log": relationship(
            MeasurementLog,
            foreign_keys=[some_event_table.c.measurement_log_before_id],
            back_populates="starts_event",
        ),
        "after_log": relationship(
            MeasurementLog,
            foreign_keys=[some_event_table.c.measurement_log_after_id],
            back_populates="ends_event",
        ),
    },
)


@pytest.fixture
def test_db():
    db_url = "./test.db"
    Path(db_url).unlink(missing_ok=True)
    engine = create_engine(f"sqlite:///{db_url}", echo=True)
    try:
        metadata.create_all(bind=engine)
        session_factory = sessionmaker(bind=engine)
        with session_factory() as session:
            yield session
    finally:
        Path(db_url).unlink(missing_ok=True)


def test_some_event(test_db: Session):
    log_entry1 = MeasurementLog(
        recorded_at=datetime.now(UTC) - timedelta(minutes=45), measurement=123
    )
    test_db.add(log_entry1)
    log_entry2 = MeasurementLog(
        recorded_at=datetime.now(UTC) - timedelta(minutes=33), measurement=123
    )
    test_db.add(log_entry2)
    test_db.commit()
    some_event = SomeEvents(
        measurement_log_before_id=log_entry1.id,
        measurement_log_after_id=log_entry2.id,
        consumer="alice",
        event="some event",
    )
    assert log_entry1.id
    assert log_entry2.id
    test_db.add(some_event)
    test_db.commit()

已尝试的解决方法

将SomeEvents对象切换为dataclass或普通类
将id字段设置为None而非init=False
查阅SQLAlchemy命令式映射文档

使用版本

attrs==25.3.0
sqlalchemy==2.0.40

解决方案

核心问题

问题出在属性映射的优先级：当你同时定义了外键字段（measurement_log_before_id、measurement_log_after_id）和对应的关系属性（before_log、after_log）时，SQLAlchemy默认会优先使用关系属性来同步外键值，但你的SomeEvents实例中before_log和after_log初始为None，导致ORM覆盖了你手动设置的外键ID，最终插入时外键值变为None触发约束错误。

修复方法

有两种可行的修复方式，选择其一即可：

方式1：直接使用关系属性关联对象

不需要手动设置外键ID，直接通过关系属性绑定已持久化的MeasurementLog实例，SQLAlchemy会自动处理外键值的填充：

# 修改test_some_event函数中的SomeEvents创建代码
some_event = SomeEvents(
    before_log=log_entry1,  # 直接关联对象
    after_log=log_entry2,
    consumer="alice",
    event="some event",
)

方式2：显式指定外键字段的映射规则

在命令式映射时，告诉SQLAlchemy优先使用你手动设置的外键字段值，而非关系属性。修改SomeEvents的映射代码，添加column_property来明确映射外键列：

from sqlalchemy.orm import column_property

# 修改SomeEvents的map_imperatively部分
mapper_registry.map_imperatively(
    SomeEvents,
    some_event_table,
    properties={
        "measurement_log_before_id": column_property(some_event_table.c.measurement_log_before_id),
        "measurement_log_after_id": column_property(some_event_table.c.measurement_log_after_id),
        "before_log": relationship(
            MeasurementLog,
            foreign_keys=[some_event_table.c.measurement_log_before_id],
            back_populates="starts_event",
        ),
        "after_log": relationship(
            MeasurementLog,
            foreign_keys=[some_event_table.c.measurement_log_after_id],
            back_populates="ends_event",
        ),
    },
)

这样SQLAlchemy会正确识别你手动设置的外键ID值，不会被None的关系属性覆盖。

额外说明

在DDD实践中，推荐使用方式1，因为直接操作领域对象关系更符合领域模型的设计思想，避免手动处理外键ID这类底层数据库细节，让ORM负责数据持久化的映射工作。

内容的提问来源于stack exchange，提问作者5th

使用attrs+SQLAlchemy命令式映射插入外键时触发非空约束错误

问题 #

可复现代码 #

已尝试的解决方法 #

使用版本 #

解决方案 #

核心问题 #

修复方法 #