How to set cascade option in sqlmodel, I want to delete all the childrens when parent is deleted

[basiltt]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the SQLModel documentation, with the integrated search.
• I already searched in Google "How to X in SQLModel" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to SQLModel but to Pydantic.
• I already checked if it is not related to SQLModel but to SQLAlchemy.

Commit to Help

• I commit to help with one of those options 👆

Example Code

class User(UserBase, table=True):
    id: int = Field(primary_key=True, index=True)
    created_datetime: datetime = Field(sa_column=Column(DateTime(timezone=True), nullable=False, default=func.now()))
    is_phone_verified: Optional[bool] = Field(default=False)
    is_email_verified: Optional[bool] = Field(default=False)
    is_password_reset_pending: Optional[bool] = Field(default=False)
    addresses: List["UserAddress"] = Relationship(back_populates="useraddress")


class UserAddress(UserAddressBase, table=True):
    id: int = Field(primary_key=True, index=True)
    user_id: int = Field(foreign_key="user.id", nullable=False, index=True)
    is_primary_address: Optional[bool] = Field(default=True)
    geo_location: Optional[str] = Field(
        sa_column=Column(Geometry(geometry_type='POINT', srid=4326), nullable=True))
    added_datetime: datetime = Field(sa_column=Column(DateTime(timezone=True), nullable=False, default=func.now()))
    team: Optional[User] = Relationship(back_populates="addresses")

Description

I couldn't find any samples to delete the children's of a parent when it is deleted.

Operating System

Windows

Operating System Details

Windows 10 20H2

SQLModel Version

sqlmodel==0.0.6

Python Version

Python 3.9.9

Additional Context

No response

[basiltt]

Please find the updated example code below

class User(UserBase, table=True):
    id: int = Field(primary_key=True, index=True)
    created_datetime: datetime = Field(sa_column=Column(DateTime(timezone=True), nullable=False, default=func.now()))
    is_phone_verified: Optional[bool] = Field(default=False)
    is_email_verified: Optional[bool] = Field(default=False)
    is_password_reset_pending: Optional[bool] = Field(default=False)
    addresses: List["UserAddress"] = Relationship(back_populates="user_address")


class UserAddress(UserAddressBase, table=True):
    id: int = Field(primary_key=True, index=True)
    user_id: int = Field(foreign_key="user.id", nullable=False, index=True)
    is_primary_address: Optional[bool] = Field(default=True)
    geo_location: Optional[str] = Field(
        sa_column=Column(Geometry(geometry_type='POINT', srid=4326), nullable=True))
    added_datetime: datetime = Field(sa_column=Column(DateTime(timezone=True), nullable=False, default=func.now()))
    user_address: Optional[User] = Relationship(back_populates="addresses")

[basiltt]

I have managed to achieve this by using the legacy SQLAlchemy relationship features by passing the sa_relationship parameter.

class User(UserBase, table=True):
    id: int = Field(primary_key=True, index=True)
    created_datetime: datetime = Field(sa_column=Column(DateTime(timezone=True), nullable=False, default=func.now()))
    is_phone_verified: Optional[bool] = Field(default=False)
    is_email_verified: Optional[bool] = Field(default=False)
    is_password_reset_pending: Optional[bool] = Field(default=False)
    addresses: List["UserAddress"] = Relationship(
        sa_relationship=relationship("UserAddress", cascade="all, delete", back_populates="user_address"))


class UserAddress(UserAddressBase, table=True):
    id: int = Field(primary_key=True, index=True)
    user_id: int = Field(foreign_key="user.id", nullable=False, index=True)
    is_primary_address: Optional[bool] = Field(default=True)
    geo_location: Optional[str] = Field(
        sa_column=Column(Geometry(geometry_type='POINT', srid=4326), nullable=True))
    added_datetime: datetime = Field(sa_column=Column(DateTime(timezone=True), nullable=False, default=func.now()))
    user_address: Optional[User] = Relationship(sa_relationship=relationship("User", back_populates="addresses"))

[JLHasson]

You can pass the cascade delete parameters through sa_relationship_kwargs:

class User(UserBase, table=True):
    addresses: List["UserAddress"] = Relationship(sa_relationship_kwargs={"cascade": "delete"}, back_populates="user_address")

Granted, I wasn't able to get it to work with {"cascade": "all, delete"}. Full reproduce code:

from typing import List, Optional

from sqlmodel import Field, Relationship, Session, SQLModel, create_engine


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    tests: List["Test"] = Relationship(
        sa_relationship_kwargs={"cascade": "delete"}
        # Doesn't work vvv
        # sa_relationship_kwargs={"cascade": "all, delete"}
    )


class Test(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    hero_id: Optional[int] = Field(default=None, foreign_key="hero.id")


engine = create_engine("sqlite:///", echo=True)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


def create_heroes():
    hero = Hero(name="Deadpond", secret_name="Dive Wilson")
    t1 = Test()
    t2 = Test()
    hero.tests = [t1, t2]

    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.query(Hero).delete()
        session.commit()
        heroes = session.query(Hero).all()
        print(heroes)
        assert len(heroes) == 0
        tests = session.query(Test).all()
        print(tests)
        assert len(tests) == 0


def main():  #
    create_db_and_tables()  #
    create_heroes()  #


if __name__ == "__main__":  #
    main()

"""
2022-01-13 15:37:54,273 INFO sqlalchemy.engine.Engine DELETE FROM hero
2022-01-13 15:37:54,273 INFO sqlalchemy.engine.Engine [generated in 0.00006s] ()
2022-01-13 15:37:54,273 INFO sqlalchemy.engine.Engine COMMIT
2022-01-13 15:37:54,273 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2022-01-13 15:37:54,274 INFO sqlalchemy.engine.Engine SELECT hero.id AS hero_id 
FROM hero
2022-01-13 15:37:54,274 INFO sqlalchemy.engine.Engine [generated in 0.00006s] ()
[]
2022-01-13 15:37:54,275 INFO sqlalchemy.engine.Engine SELECT test.id AS test_id, test.hero_id AS test_hero_id 
FROM test
2022-01-13 15:37:54,275 INFO sqlalchemy.engine.Engine [generated in 0.00006s] ()
[]
2022-01-13 15:37:54,275 INFO sqlalchemy.engine.Engine ROLLBACK
"""

[JLHasson]

Following up on my response above. For SQLite engines foreign key constraints are disabled by default. SQLAlchemy provides the ability to enable this per connection via an event listener documented here: https://docs.sqlalchemy.org/en/13/dialects/sqlite.html#foreign-key-support

I believe what I posted above is likely not the correct way to implement this. Instead, I believe you should set the constraint on the column instead of the relationship which will use the DB to resolve the constraint as opposed to the ORM. You should then set the cascade parameters on the Relationship to instruct the ORM how to handle deletes with local objects (documented here: https://docs.sqlalchemy.org/en/14/orm/cascades.html#using-foreign-key-on-delete-cascade-with-orm-relationships).

Therefore the updated code should read something like the following:

# SQLite specific
from sqlalchemy.engine import Engine
from sqlalchemy import event
from sqlite3 import Connection as SQLite3Connection


@event.listens_for(Engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
    if isinstance(dbapi_connection, SQLite3Connection):
        cursor = dbapi_connection.cursor()
        cursor.execute("PRAGMA foreign_keys=ON")
        cursor.close()

# Update models
class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    tests: List["Test"] = Relationship(
        sa_relationship_kwargs={
            "cascade": "all, delete",  # Instruct the ORM how to track changes to local objects
        },
    )


class Test(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    hero_id: Optional[int] = Field(
        sa_column=Column(Integer, ForeignKey("hero.id", ondelete="CASCADE"))  # Set the foreign key behavior on the table metadata
    )

[Zaffer]

@JLHasson why does your relationship not have a back_populates kwarg?

[JLHasson]

As far as I know it's an optional parameter, you just need it if you want the value to be filled on the object the relation is to. You should be able to add it here I believe without issue?

[FilipeMarch]

Hello, guys, I have a Worker that can have multiple Transaction (each transaction has a worker_id foreign_key).

When I delete the Worker whose ID is 1, I would like that each Transaction that was pointing to this worker would have Transaction.worker_id == None. But executing this code, the transaction is still pointing to a non existing worker.

from typing import Optional
from sqlmodel import Field, SQLModel,create_engine, Session


class Worker(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)


class Transaction(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    worker_id: int = Field(index=True, foreign_key="worker.id")

        
sqlite_file_name = "database/database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

engine = create_engine(sqlite_url, echo=False)
SQLModel.metadata.create_all(engine)

# Create a worker
with Session(engine) as session:
    worker = Worker(name='Worker 1')
    session.add(worker)
    session.commit()
    session.refresh(worker)

# Create a transaction pointing to this worker
with Session(engine) as session:
    transaction = Transaction(worker_id=worker.id)
    session.add(transaction)
    session.commit()
    session.refresh(transaction)

# Delete the worker
with Session(engine) as session:
    session.delete(worker)
    session.commit()
    
# Print all transactions on database
with Session(engine) as session:
    transactions = session.query(Transaction).all()
    print(transactions)

>>> [Transaction(worker_id=1, id=1)]

What I was expecting:

>>> [Transaction(worker_id=None, id=1)]

Does anyone know why this transaction is still pointing to worker_id == 1 and how to make it to automatically point to None when I delete the Worker whose ID is 1?

[FilipeMarch]

I have achieved the behaviour I was looking for by creating a Relationship in Worker like this:

class Worker(SQLModel, table=True):
   id: Optional[int] = Field(default=None, primary_key=True)
   name: str = Field(index=True)
   transactions: List["Transaction"] = Relationship(
       sa_relationship_kwargs={
           "cascade": ""
       }
   )

Also, worker_id must become Optional, because it will become None once the worker is deleted.

class Transaction(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    worker_id: Optional[int] = Field(index=True, foreign_key="worker.id")

The whole code becomes:

from typing import Optional, List
from sqlmodel import Field, SQLModel, create_engine, Session, Relationship



class Worker(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    transactions: List["Transaction"] = Relationship(
        sa_relationship_kwargs={
            "cascade": ""
        }
    )


class Transaction(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    worker_id: Optional[int] = Field(index=True, foreign_key="worker.id")

        
sqlite_file_name = "database/database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

engine = create_engine(sqlite_url, echo=False)
SQLModel.metadata.create_all(engine)

# Create a worker
with Session(engine) as session:
    worker = Worker(name='Worker 1')
    session.add(worker)
    session.commit()
    session.refresh(worker)

# Create a transaction pointing to this worker
with Session(engine) as session:
    transaction = Transaction(worker_id=worker.id)
    session.add(transaction)
    session.commit()
    session.refresh(transaction)

# Delete the worker
with Session(engine) as session:
    session.delete(worker)
    session.commit()
    
# Print all transactions on database
with Session(engine) as session:
    transactions = session.query(Transaction).all()
    print(transactions)

The output:

>>> [Transaction(worker_id=None, id=1)]

[hp0404]

worked by setting sa_relationship_kwargs={"cascade": "all,delete,delete-orphan"}

[shintoo]

For anyone else struggling with this, recommend pinning sqlalchemy to 1.4.35 per #315. With the definitions of #213 (comment) and this sqlalchemy version I was able to resolve my issue.

[GiorgioSgl]

Why not implement something like this and avoid use SQLAlchemy?

class MyModel(SQLModel, table=True):
    id: str
    fk_id: str = Field(
        foreign_key=f"{SchemaName.non_performing_loans.value}.{tenant_name}.tenant_id",
        default=None,
        description="Tenant id",
        delete="cascade"
    )