I am building a polymorphic relationship between classes A and B through a Link class.
I want to achieve that when an A object is deleted, the linked B objects are deleted too.
To do that, I defined a relationship between A and Link, with a cascade relationship all, delete
, in this way as soon as A is deleted we will try to delete all the links, and then I intercepted the delete on the Link using an event listener before_delete
class and I used it to delete also the B object.
A better solution can exist, and most of all I am concerned about the connection object in the event listener that doesn’t seem to be bound to any session.
class A(BaseORMModel):
id: Mapped[int] = mapped_column(
primary_key=True,
)
links = relationship(
"Link",
primaryjoin="and_("
"foreign(Link.source_id)==A.id,"
"Link.source_type=='TypeA'"
")",
cascade="all, delete",
)
class B(BaseORMModel):
id: Mapped[int] = mapped_column(
primary_key=True,
)
class Link(BaseOrmModel):
id: Mapped[int] = mapped_column(
primary_key=True,
)
source_id: Mapped[int] = mapped_column(
index=True,
)
source_type: Mapped[str]
related_id: Mapped[int] = mapped_column(
index=True,
)
related_type: Mapped[str]
@event.listens_for(Link, "before_delete")
def delete_links(mapper, connection, Link):
if Link.related_type == 'TypeB':
q = delete(B).where(B.id == Link.related_id)
connection.execute(
q
)
print(q.compile(compile_kwargs={"literal_binds": True}))