Data Integrity: Raise error on attempt to delete an object required via a Relationship

[clstaudt]

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 Contact(SQLModel, table=True):
    """An entry in the address book."""

    id: Optional[int] = Field(default=None, primary_key=True)
    first_name: Optional[str]
    last_name: Optional[str]
    company: Optional[str]
    email: Optional[str]
    address_id: Optional[int] = Field(default=None, foreign_key="address.id")
    address: Optional[Address] = Relationship(
        back_populates="contacts", sa_relationship_kwargs={"lazy": "subquery"}
    )
    invoicing_contact_of: List["Client"] = Relationship(
        back_populates="invoicing_contact", sa_relationship_kwargs={"lazy": "subquery"}
    )



class Client(SQLModel, table=True):
    """A client the freelancer has contracted with."""

    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(default="")
    # Client 1:1 invoicing Contact
    invoicing_contact_id: int = Field(default=None, foreign_key="contact.id")
    invoicing_contact: Contact = Relationship(
        back_populates="invoicing_contact_of",
        sa_relationship_kwargs={"lazy": "subquery"},
    )
    contracts: List["Contract"] = Relationship(
        back_populates="client", sa_relationship_kwargs={"lazy": "subquery"}
    )

Description

(As far as I know the documentation does not handle data integrity topics - please point me to the chapter if I am wrong.)

Consider these two model classes Contact and Client. To keep the integrity of the data model, I need the following behavior:

An exception is raised if there is an attempt to delete a Contact that is still the invoicing contact of an existing Client.

Does SQLModel support this, perhaps via SQLAlchemy?

Operating System

macOS

Operating System Details

No response

SQLModel Version

0.0.8

Python Version

3.10

Additional Context

No response

[meirdev]

The default behavior of SQLAlchemy when you delete Contact is to set NULL in the invoicing_contact_id field, to avoid this, you need to set passive_deletes='all' in the relationship field in the Client model:

class Contact(SQLModel, table=True):
    ...
    invoicing_contact_of: List["Client"] = Relationship(back_populates="invoicing_contact", sa_relationship_kwargs={"lazy": "subquery", "passive_deletes": "all"})

And now you get an error when you try to delete a Contact that has a relationship with Client:

sqlalchemy.exc.IntegrityError: (psycopg2.errors.ForeignKeyViolation) update or delete on table "contact" violates foreign key constraint "client_invoicing_contact_id_fkey" on table "client"
DETAIL:  Key (id)=(1) is still referenced from table "client".

https://docs.sqlalchemy.org/en/20/orm/relationship_api.html#sqlalchemy.orm.relationship.params.passive_deletes

[clstaudt]

@meirdev Thanks for the suggestion, it was quite difficult to find documentation on this.

However, I am not getting the desired behavior.

I have modified the model as follows:

class Contact(SQLModel, table=True):
    """An entry in the address book."""

    id: Optional[int] = Field(default=None, primary_key=True)
    first_name: Optional[str]
    last_name: Optional[str]
    company: Optional[str]
    email: Optional[str]
    address_id: Optional[int] = Field(default=None, foreign_key="address.id")
    address: Optional[Address] = Relationship(
        back_populates="contacts", sa_relationship_kwargs={"lazy": "subquery"}
    )
    invoicing_contact_of: List["Client"] = Relationship(
        back_populates="invoicing_contact", 
        sa_relationship_kwargs={"lazy": "subquery", "passive_deletes": "all"}
    )

Now if I understand correctly, this function is supposed to raise an IntegrityError if invoicing_contact_of is not empty:

    def delete_by_id(self, entity_type: Type[sqlmodel.SQLModel], entity_id: int):
        """Deletes the entity of the given type with the given id from the database"""
        logger.debug(f"deleting {entity_type} with id={entity_id}")
        with self.create_session() as session:
            session.exec(
                sqlmodel.delete(entity_type).where(entity_type.id == entity_id)
            )
            session.commit()

It doesn't, the deletion proceeds and the respective Client.invoicing_contact becomes missing.

[meirdev]

Works for me:

from typing import Optional, List

from sqlmodel import (
    Field,
    Relationship,
    SQLModel,
    create_engine,
    Session,
    delete,
)


class Contact(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    invoicing_contact_of: List["Client"] = Relationship(
        back_populates="invoicing_contact",
        sa_relationship_kwargs={"lazy": "subquery", "passive_deletes": "all"},
    )


class Client(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    invoicing_contact_id: int = Field(default=None, foreign_key="contact.id")
    invoicing_contact: Contact = Relationship(
        back_populates="invoicing_contact_of",
        sa_relationship_kwargs={"lazy": "subquery"},
    )


engine = create_engine("postgresql://postgres:postgrespw@localhost:55000", echo=True)

SQLModel.metadata.drop_all(engine)
SQLModel.metadata.create_all(engine)

with Session(engine) as session:
    contact = Contact()
    client = Client(invoicing_contact=contact)

    session.add(client)
    session.commit()
    session.refresh(client)

    session.exec(delete(Contact).where(Contact.id == 1))
    session.commit()

[clstaudt]

Strange. Will run the minimal example to investigate. But does it matter that you are using PostgreSQL and I am using SQLite?

[meirdev]

Indeed! in SQLite the foreign key constraints are disabled by default (https://www.sqlite.org/foreignkeys.html, 2), you have to enable them manually. the simplest way is to listen to the connect event and enable this option:

from sqlalchemy import event

engine = create_engine("sqlite:///")
event.listen(engine, "connect", lambda c, _: c.execute("PRAGMA foreign_keys = ON"))

[clstaudt]

@meirdev Now the IntegrityError is raised. However, the logic is bidirectional:

If I try to delete a Client I get also an IntegrityError. But this should be allowed.

2023-01-20 10:08:07.075 | ERROR    | core.intent_result:log_message_if_any:44 - (sqlite3.IntegrityError) FOREIGN KEY constraint failed
[SQL: DELETE FROM client WHERE client.id = ?]
[parameters: (1,)]
(Background on this error at: https://sqlalche.me/e/14/gkpj)
NoneType: None

[meirdev]

I think the error came from something else, in my sample code if I try to delete Client there are no errors.

session.exec(delete(Client).where(Client.id == 1))
session.commit()

[clstaudt]

@meirdev I can verify that our minimal examples are working fine. However, in the context of the entire data model, this is not working.

Basic idea of the data model: Project requires Contract requires Client required Contact.

Trying to delete a Client raises

sqlalchemy.exc.IntegrityError: (sqlite3.IntegrityError) FOREIGN KEY constraint failed
[SQL: DELETE FROM client WHERE client.id = ?]
[parameters: (1,)]
(Background on this error at: https://sqlalche.me/e/14/gkpj)

and so does trying to delete a Contract or Project.

It seems that by adding just this to the Contact model:

invoicing_contact_of: List["Client"] = Relationship(
    back_populates="invoicing_contact",
    sa_relationship_kwargs={
        "lazy": "subquery",
        "passive_deletes": "all",  # can't delete contact if it's used in a client
    },
)

... I have made every object with a path to an existing Contact undeletable.

The "FOREIGN KEY contraint failed" message is also unhelpful because it doesn't tell us any details. Is there a way to get more info?

[YuriiMotov]

(As far as I know the documentation does not handle data integrity topics - please point me to the chapter if I am wrong.)

There is a dedicated section in current version of docs: Cascade Delete Relationships

@clstaudt, do you think we can close this issue now?

[YuriiMotov]

This section of docs describes exactly what is needed: https://sqlmodel.tiangolo.com/tutorial/relationship-attributes/cascade-delete-relationships/#ondelete-with-restrict

from typing import List, Optional

from sqlalchemy.exc import IntegrityError
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, delete, text


class Contact(SQLModel, table=True):
    """An entry in the address book."""

    id: Optional[int] = Field(default=None, primary_key=True)
    ...
    invoicing_contact_of: List["Client"] = Relationship(
        back_populates="invoicing_contact",
        sa_relationship_kwargs={"lazy": "subquery"},
        passive_deletes="all",
    )


class Client(SQLModel, table=True):
    """A client the freelancer has contracted with."""

    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(default="")
    # Client 1:1 invoicing Contact
    invoicing_contact_id: int = Field(
        default=None, foreign_key="contact.id", ondelete="RESTRICT"
    )
    invoicing_contact: Contact = Relationship(
        back_populates="invoicing_contact_of",
        sa_relationship_kwargs={"lazy": "subquery"},
    )
    ...


engine = create_engine("sqlite:///")


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)
    with engine.connect() as connection:
        connection.execute(text("PRAGMA foreign_keys=ON"))  # for SQLite only


def create_data():
    with Session(engine) as session:
        clients = [Client(name="Client 1"), Client(name="Client 2")]
        session.add(Contact(invoicing_contact_of=clients))
        session.commit()


def main():
    create_db_and_tables()
    create_data()

    with Session(engine) as session:
        contact = session.get(Contact, 1)

        # 1. Ensure Contact can't be deleted if it's linked to clients
        try:
            session.delete(contact)
            session.commit()
        except IntegrityError:
            print("1. Can't delete contact linked to clients")
        else:
            raise RuntimeError("1. Something wrong...")

    with Session(engine) as session:
        # 2. Ensure clients can be deleted even if linked to Contact
        session.exec(delete(Client))
        session.commit()
        print("2. Deleted clients successfully")

    with Session(engine) as session:
        # 3. Ensure contact is still in the DB
        contact = session.get(Contact, 1)
        assert contact is not None
        print("3. Contact is still in DB")

        # 4. Ensure Contact now can be deleted
        session.delete(contact)
        session.commit()
        print("4. Deleted contact successfully")


if __name__ == "__main__":
    main()