Question Details

No question body available.

Tags

python postgresql sqlalchemy many-to-many

Answers (2)

April 7, 2025 Score: 2 Rep: 21 Quality: Low Completeness: 60%

To ensure the position is set correctly when committing the objects is by hooking into the beforeinsert event for the DbAssessmentsTasks model. You can then automatically calculate and set the position before the insert occurs.

Something like this:

from sqlalchemy import event

@event.listensfor(DbAssessmentsTasks, "beforeinsert") def setposition(mapper, connection, target): if isinstance(target, DbAssessmentsTasks): position = target.position or 0 if position == 0: result = connection.execute( text(f"SELECT COALESCE(MAX(position), 0) + 1 FROM assessmenttasks WHERE assessmentid = :assessmentid"), {"assessmentid": target.assessmentid} ) target.position = result.scalar()

This code listens for the beforeinsert event on DbAssessmentsTasks and sets the position automatically by querying the current highest position in the table for the associated assessment_id. It then increments it by one before the insert occurs.

April 7, 2025 Score: 1 Rep: 545 Quality: Low Completeness: 60%

Define DbAssessment.taskslink for association table

1.- Add a new relationship to DbAssessment that reflects the association table directly:

class DbAssessment(DbBase): tablename = "assessments"

taskslink: Mapped[list["DbAssessmentsTasks"]] = relationship(backpopulates="assessment", cascade="all, delete-orphan") tasks: Mapped[list["DbTask"]] = relationship(secondary="assessmenttasks", backpopulates="assessments")

2.- And in DbAssessmentsTasks:

class DbAssessmentsTasks(DbBase): tablename = "assessment
tasks" tableargs = (UniqueConstraint("assessmentid", "position"),)

@declaredattr def id(cls): return None

position: Mapped[int] = mappedcolumn(Integer, nullable=False) assessmentid: Mapped[UUID] = mappedcolumn(ForeignKey("assessments.id"), primarykey=True) taskid: Mapped[UUID] = mappedcolumn(ForeignKey("tasks.id"), primarykey=True)

assessment = relationship("DbAssessment", backpopulates="taskslink") task = relationship("DbTask")

3. Add an event listener to populate DbAssessmentsTasks:

from sqlalchemy import event

@event.listensfor(sessionmaker(bind=engine), "beforeflush") def populateassociationpositions(session, flushcontext, instances): for obj in session.new: if isinstance(obj, DbAssessment) and obj.tasks: obj.taskslink.clear() for i, task in enumerate(obj.tasks, start=1): obj.taskslink.append(DbAssessmentsTasks( task=task, position=i ))

Then you can just do:

db
assessment = DbAssessment( tasks=[DbPrimer(), DbExercise(), DbExercise()] ) session.add(db_assessment) session.commit()