[Python] Dataclass/attrs based orm models and their relationships

Discussão em 'Python' iniciado por Stack, Maio 5, 2025.

  1. Stack

    Stack Membro Participativo

    I am building a small python application to learn Domain-Driven-Design (DDD) approaches. Therefore I am using a dataclass/attrs class as my domain model and also use this class to imperatively model my orm instances. However I encounter a bug when I try to insert a class containing foreign keys:

    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)]
    (Background on this error at: https://sqlalche.me/e/20/gkpj) [123, 1]


    The odd thing is that the attrs/dataclass has the id value. But the orm class looses it somehow during the session.commit()

    To help reproduce the issue I will add a rather long minimal viable example which allows you to reproduce the issue at the end. Feel free to switch pytest out and run it as a script instead.

    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()


    I tried:


    I am using:

    attrs==25.3.0
    sqlalchemy==2.0.40

    Continue reading...

Compartilhe esta Página