[Python] How to automatically update extra column in sqlalchemy association table?

I have a postgresql database setup with sqlalchemy, joined table inheritance and a many-to-many association table with an extra column position, i.e. assessments and tasks are related via the assessments_tasks association table and this table has the extra column.

If I now want to add an assessment, which can have multiple tasks like e.g. one primer and two exercises, I want that the position in the association table is automatically written according to the position in the tasks list.

DbAssessment(tasks=[DbPrimer(), DbExercise(), DbExercise()])


This should insert one entry in DbAssessment, one entry in DbPrimer, two entries in DbExercise, three entries in DbTask and these three tasks should be in the DbAssessmentsTasks with following positions (automatically calculated):

--------------------------------------------
| position | assessment_id | task_id |
--------------------------------------------
| 1 | assessment.id | primer_1.id |
| 2 | assessment.id | exercise_1.id |
| 3 | assessment.id | exercise_2.id |
--------------------------------------------


The problem is, that I am stuck at getting the positions automatically calculated and inserted in the association table. I want that the business logic is completely in the hand of sqlalchemy or postgresql. I tried using an AssociationProxy, but I have problems to understand how the toy example can be used in my case and also how to get the creator working. I hope that I can get at least some help which gets me in the right direction.

Here is a minimal example, which can be run out-of-the-box after installing sqlalchemy testcontainers psycopg2:

from uuid import UUID

from sqlalchemy import ForeignKey, Integer, String, UniqueConstraint, Uuid, create_engine, text
from sqlalchemy.orm import (
DeclarativeBase, Mapped, declared_attr, mapped_column, relationship, sessionmaker
)
from testcontainers.postgres import PostgresContainer


class DbBase(DeclarativeBase):
id: Mapped[UUID] = mapped_column(Uuid, primary_key=True, server_default=text("gen_random_uuid()"))


class DbAssessment(DbBase):
__tablename__ = "assessments"

tasks: Mapped[list["DbTask"]] = relationship(secondary="assessment_tasks", back_populates="assessments")


class DbTask(DbBase):
__tablename__ = "tasks"
__mapper_args__ = {"polymorphic_on": "task_type", "polymorphic_identity": "task"}

assessments: Mapped[list["DbAssessment"]] = relationship(secondary="assessment_tasks", back_populates="tasks")
task_type: Mapped[str] = mapped_column(String(length=8), nullable=False)


class DbPrimer(DbTask):
__tablename__ = "primers"
__mapper_args__ = {"polymorphic_identity": "primer"}

id: Mapped[UUID] = mapped_column(ForeignKey("tasks.id"), primary_key=True, nullable=False)


class DbExercise(DbTask):
__tablename__ = "exercises"
__mapper_args__ = {"polymorphic_identity": "exercise"}

id: Mapped[UUID] = mapped_column(ForeignKey("tasks.id"), primary_key=True, nullable=False)


class DbAssessmentsTasks(DbBase):
__tablename__ = "assessment_tasks"
__table_args__ = (UniqueConstraint("assessment_id", "position"),)

@declared_attr
def id(cls): return None

position: Mapped[int] = mapped_column(Integer, nullable=False)
assessment_id: Mapped[UUID] = mapped_column(ForeignKey("assessments.id"), primary_key=True)
task_id: Mapped[UUID] = mapped_column(ForeignKey("tasks.id"), primary_key=True)


if __name__ == "__main__":
with PostgresContainer("postgres:latest") as postgres:
engine = create_engine(postgres.get_connection_url())
DbBase.metadata.create_all(bind=engine)

db_assessment = DbAssessment(
tasks=[DbPrimer(), DbExercise(), DbExercise()]
)
with sessionmaker(bind=engine)() as session:
session.add(db_assessment)
session.commit()


☝️should throw the expected error:

sqlalchemy.exc.IntegrityError: (psycopg2.errors.NotNullViolation) null value in column "position" of relation "assessment_tasks" violates not-null constraint
​

How can I instruct sqlalchemy or postgresql to automatically insert the calculated positions (probably something similar to [p for p, _ in enumerate(db_assessment.tasks, start=1)]) into the many-to-many association table?

Continue reading...