I have a simple SqlAlchemy Relationship defined as follows
class User(BaseModel):
cars = relationship(
"Car",
back_populates="user",
passive_deletes="all",
)
class Car(BaseModel):
user_id: Mapped[Optional[uuid.UUID]] = mapped_column(
ForeignKey("user_id", ondelete="CASCADE"),
index=True,
)
My test is run as follows
@staticmethod
async def test_car_is_cascade_deleted(
db, fixture_user, fixture_car
):
user = await fixture_user(). # just a factory that creates a User object and FLUSH
car = await fixture_car(
user=user
)
await db.refresh(car)
await db.delete(user)
await db.commit()
assert await db.get(User, user.id) is None
assert await db.get(car, car.id) is None # this fails if await db.refresh(car) is not called
As described in the comment, the await db.refresh(car)
looks needed to make the
assert await db.get(car, car.id)
getting a None value, otherwise an actual object is retrieved.
I also echoed the queries in SqlAlchemy and I clearly see the Delete instructions
DELETE FROM car WHERE car.user_id = %(user_id)s::UUID
2024-07-18 17:48:46,521 INFO sqlalchemy.engine.Engine [generated in 0.00007s] [{'user_id': 'b652d7c5-6328-4725-b276-914264c0064a'}, {'user_id': 'b652d7c5-6328-4725-b276-914264c0064a'}]
2024-07-18 17:48:46,523 INFO sqlalchemy.engine.Engine DELETE FROM user WHERE user.id = %(id)s::UUID {'id': 'b652d7c5-6328-4725-b276-914264c0064a'}
What I didn’t get from SqlAlchemy session management?
1