관계 삭제 연쇄하기

관계 삭제 연쇄하기 (Cascade Delete Relationships)

히어로와 관계를 맺고 있는 팀을 삭제하면 어떻게 될까요? 그 히어로들도 자동으로 함께 삭제되는 게 맞을까요? 이걸 "cascade(연쇄)"라고 불러요. 첫 번째 삭제가 다른 삭제들을 연쇄적으로 불러일으키기 때문이죠. 아니면 데이터베이스에서 그들의 team_id를 NULL로 바꿔야 할까요? SQLModel에서 이를 어떻게 설정하는지 함께 살펴볼게요.

출처: 공식문서

📌 참고

이 기능은 cascade_delete, ondelete, passive_deletes를 포함해 SQLModel 0.0.21 버전부터 사용할 수 있어요.

처음의 히어로와 팀

우리에게 이런 team과 hero가 있다고 해볼게요.

Team Table

id	name	headquarters
1	Z-Force	Sister Margaret's Bar
2	Preventers	Sharp Tower
3	Wakaland	Wakaland Capital City

Hero Table

id	name	secret_name	age	team_id
1	Deadpond	Dive Wilson		1
2	Rusty-Man	Tommy Sharp	48	2
3	Spider-Boy	Pedro Parqueador		2
4	Black Lion	Trevor Challa	35	3
5	Princess Sure-E	Sure-E		3

히어로가 있는 팀 삭제하기

팀을 삭제할 때는 연결된 히어로들을 어떻게 처리할지 정해야 해요.

기본적으로는 팀을 가리키는 그들의 외래 키가 데이터베이스에서 NULL로 설정돼요.

하지만 우리가 연결된 히어로들도 자동으로 삭제되길 원한다고 해볼게요.

예를 들어 Wakaland 팀을 삭제한다고 하면:

우리는 Black Lion과 Princess Sure-E 히어로들도 자동으로 삭제되길 원해요.

그러면 결과적으로 이렇게 team들과 hero들이 남게 되겠죠.

자동 삭제 설정하기

이 자동 삭제는 두 곳에서 설정해요:

Python 코드에서
데이터베이스에서

Python에서 cascade_delete로 삭제하기

Relationship()을 만들 때 cascade_delete=True로 설정할 수 있어요.

이 설정은 SQLModel이 처음 레코드(팀)가 삭제될 때 관련 레코드(히어로들)를 자동으로 삭제하도록 해줘요.

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


class Team(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    headquarters: str

    heroes: list["Hero"] = Relationship(back_populates="team", cascade_delete=True)

# Code below omitted 👇

👀 전체 파일 미리보기 (Full file preview)

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


class Team(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    headquarters: str

    heroes: list["Hero"] = Relationship(back_populates="team", cascade_delete=True)


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)

    team_id: int | None = Field(default=None, foreign_key="team.id", ondelete="CASCADE")
    team: Team | None = Relationship(back_populates="heroes")


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

engine = create_engine(sqlite_url, echo=True)


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


def create_heroes():
    with Session(engine) as session:
        team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
        team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")

        hero_deadpond = Hero(
            name="Deadpond", secret_name="Dive Wilson", team=team_z_force
        )
        hero_rusty_man = Hero(
            name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers
        )
        hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
        session.add(hero_deadpond)
        session.add(hero_rusty_man)
        session.add(hero_spider_boy)
        session.commit()

        session.refresh(hero_deadpond)
        session.refresh(hero_rusty_man)
        session.refresh(hero_spider_boy)

        print("Created hero:", hero_deadpond)
        print("Created hero:", hero_rusty_man)
        print("Created hero:", hero_spider_boy)

        hero_spider_boy.team = team_preventers
        session.add(hero_spider_boy)
        session.commit()
        session.refresh(hero_spider_boy)
        print("Updated hero:", hero_spider_boy)

        hero_black_lion = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
        hero_sure_e = Hero(name="Princess Sure-E", secret_name="Sure-E")
        team_wakaland = Team(
            name="Wakaland",
            headquarters="Wakaland Capital City",
            heroes=[hero_black_lion, hero_sure_e],
        )
        session.add(team_wakaland)
        session.commit()
        session.refresh(team_wakaland)
        print("Team Wakaland:", team_wakaland)


def delete_team():
    with Session(engine) as session:
        statement = select(Team).where(Team.name == "Wakaland")
        team = session.exec(statement).one()
        session.delete(team)
        session.commit()
        print("Deleted team:", team)


def select_deleted_heroes():
    with Session(engine) as session:
        statement = select(Hero).where(Hero.name == "Black Lion")
        result = session.exec(statement)
        hero = result.first()
        print("Black Lion not found:", hero)

        statement = select(Hero).where(Hero.name == "Princess Sure-E")
        result = session.exec(statement)
        hero = result.first()
        print("Princess Sure-E not found:", hero)


def main():
    create_db_and_tables()
    create_heroes()
    delete_team()
    select_deleted_heroes()


if __name__ == "__main__":
    main()

이 설정을 쓰면, 우리가 팀을 삭제할 때 SQLModel(정확히는 SQLAlchemy)은 이렇게 동작해요:

관련 레코드의 객체들을 로드합니다. 이 경우엔 히어로들이죠. 로드되어 있지 않다면 데이터베이스에 SELECT 쿼리를 보내 가져와요.
각 관련 레코드(각 히어로)를 포함하는 DELETE 쿼리를 데이터베이스에 보내요.
마지막으로 처음 레코드(팀)를 또 다른 DELETE 쿼리로 삭제해요.

이렇게 하면 내부 Python 코드가 각각에 필요한 SQL 쿼리를 만들어 내면서 관련 레코드 삭제를 직접 처리해요.

💡 팁

cascade_delete 파라미터는 외래 키가 없는 쪽의 모델에 있는 Relationship()에 설정해요.

기술적 세부사항

Relationship()에서 cascade_delete=True로 설정하면 SQLAlchemy가 cascade="all, delete-orphan"을 사용하도록 설정돼요. 연쇄 삭제를 원할 때 가장 흔하고 유용한 설정이죠.

더 자세한 내용은 SQLAlchemy 문서에서 확인할 수 있어요.

데이터베이스에서 ondelete로 삭제하기

앞서 본 cascade_delete는 Python 코드에서 자동 삭제를 처리한다는 걸 알게 됐어요.

하지만 누군가 우리 코드를 거치지 않고 데이터베이스에 직접 접근해서 SQL로 팀을 삭제하면 어떻게 될까요?

그런 경우를 위해, 우리는 Field()의 ondelete 파라미터로 데이터베이스 자체가 관련 레코드를 자동 삭제하도록 설정할 수 있어요.

ondelete 옵션

ondelete 파라미터는 데이터베이스에서 외래 키 컬럼에 SQL ON DELETE를 설정해요.

ondelete는 이런 값을 가질 수 있어요:

CASCADE: 관련 레코드(팀)가 삭제될 때 이 레코드(히어로)를 자동으로 삭제해요.
SET NULL: 관련 레코드가 삭제될 때 이 외래 키(hero.team_id) 필드를 NULL로 설정해요.
RESTRICT: 외래 키 값이 있으면 오류를 일으켜서 이 레코드(히어로)의 삭제를 막아요.

ondelete를 CASCADE로 설정

부모가 삭제될 때 관련 레코드가 자동 삭제되도록 데이터베이스를 설정하려면 ondelete="CASCADE"를 쓰면 돼요.

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


class Team(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    headquarters: str

    heroes: list["Hero"] = Relationship(back_populates="team", cascade_delete=True)


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)

    team_id: int | None = Field(default=None, foreign_key="team.id", ondelete="CASCADE")
    team: Team | None = Relationship(back_populates="heroes")

# Code below omitted 👇

이제 데이터베이스에 테이블을 만들면, Hero 테이블의 team_id 컬럼은 정의에 ON DELETE CASCADE가 데이터베이스 수준에서 포함돼요.

이 설정은 관련 레코드(팀)가 삭제될 때 데이터베이스가 레코드(히어로들)를 자동으로 삭제하도록 해줘요.

💡 팁

ondelete 파라미터는 외래 키가 있는 쪽의 모델에 있는 Field()에 설정해요.

cascade_delete 또는 ondelete 사용하기

이쯤이면 cascade_delete를 써야 할지 ondelete를 써야 할지 궁금할 거예요. 답은: 둘 다! 🤓

ondelete는 누군가 데이터베이스에 직접 접근하는 경우를 대비해 데이터베이스를 설정해요.

하지만 cascade_delete는 SQLAlchemy에게 메모리 안의 Python 객체도 삭제해야 한다고 알려 주기 위해 여전히 필요해요.

외래 키 제약 지원

일부 데이터베이스는 외래 키 제약을 지원하지 않아요.

예를 들어 SQLite는 기본적으로 지원하지 않아요. 직접 SQL 명령으로 수동 활성화해야 하죠:

PRAGMA foreign_keys = ON;

그래서 일반적으로는 cascade_delete와 ondelete 둘 다 설정해 두는 게 좋아요.

💡 팁

SQLModel(SQLAlchemy)의 기본 자동 동작을 비활성화하고 데이터베이스에만 맡기는 방법은 아래 passive_deletes 부분에서 더 배우게 될 거예요.

Relationship()의 cascade_delete와 Field()의 ondelete

기억해 둘 메모 하나예요... 🤓

ondelete는 외래 키가 있는 Field()에 둡니다. "일대다(one-to-many)" 관계에서 "다(很多)" 쪽이죠.

class Hero(SQLModel, table=True):
    ...

    team_id: int = Field(foreign_key="team.id", ondelete="CASCADE")

cascade_delete는 Relationship()에 둡니다. 보통 "일대다" 관계에서 "일" 쪽, 즉 외래 키가 없는 쪽이에요.

class Team(SQLModel, table=True):
    ...

    heroes: list[Hero] = Relationship(cascade_delete=True)

팀과 그 히어로들 삭제하기

이제 팀을 삭제할 때 추가로 할 일은 없어요. 팀의 히어로를 자동으로 삭제하니까요.

# Code above omitted 👆

def delete_team():
    with Session(engine) as session:
        statement = select(Team).where(Team.name == "Wakaland")
        team = session.exec(statement).one()
        session.delete(team)
        session.commit()
        print("Deleted team:", team)

# Code below omitted 👇

👀 전체 파일 미리보기 (Full file preview)

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


class Team(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    headquarters: str

    heroes: list["Hero"] = Relationship(back_populates="team", cascade_delete=True)


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)

    team_id: int | None = Field(default=None, foreign_key="team.id", ondelete="CASCADE")
    team: Team | None = Relationship(back_populates="heroes")


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

engine = create_engine(sqlite_url, echo=True)


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


def create_heroes():
    with Session(engine) as session:
        team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
        team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")

        hero_deadpond = Hero(
            name="Deadpond", secret_name="Dive Wilson", team=team_z_force
        )
        hero_rusty_man = Hero(
            name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers
        )
        hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
        session.add(hero_deadpond)
        session.add(hero_rusty_man)
        session.add(hero_spider_boy)
        session.commit()

        session.refresh(hero_deadpond)
        session.refresh(hero_rusty_man)
        session.refresh(hero_spider_boy)

        print("Created hero:", hero_deadpond)
        print("Created hero:", hero_rusty_man)
        print("Created hero:", hero_spider_boy)

        hero_spider_boy.team = team_preventers
        session.add(hero_spider_boy)
        session.commit()
        session.refresh(hero_spider_boy)
        print("Updated hero:", hero_spider_boy)

        hero_black_lion = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
        hero_sure_e = Hero(name="Princess Sure-E", secret_name="Sure-E")
        team_wakaland = Team(
            name="Wakaland",
            headquarters="Wakaland Capital City",
            heroes=[hero_black_lion, hero_sure_e],
        )
        session.add(team_wakaland)
        session.commit()
        session.refresh(team_wakaland)
        print("Team Wakaland:", team_wakaland)


def delete_team():
    with Session(engine) as session:
        statement = select(Team).where(Team.name == "Wakaland")
        team = session.exec(statement).one()
        session.delete(team)
        session.commit()
        print("Deleted team:", team)


def select_deleted_heroes():
    with Session(engine) as session:
        statement = select(Hero).where(Hero.name == "Black Lion")
        result = session.exec(statement)
        hero = result.first()
        print("Black Lion not found:", hero)

        statement = select(Hero).where(Hero.name == "Princess Sure-E")
        result = session.exec(statement)
        hero = result.first()
        print("Princess Sure-E not found:", hero)


def main():
    create_db_and_tables()
    create_heroes()
    delete_team()
    select_deleted_heroes()


if __name__ == "__main__":
    main()

히어로들이 삭제됐는지 확인하기

Wakaland 팀을 삭제한 뒤에 Black Lion과 Princess Sure-E 히어로들도 삭제됐는지 확인해 볼게요.

데이터베이스에서 그들을 조회하려 하면 더 이상 찾을 수 없을 거예요.

# Code above omitted 👆

def select_deleted_heroes():
    with Session(engine) as session:
        statement = select(Hero).where(Hero.name == "Black Lion")
        result = session.exec(statement)
        hero = result.first()
        print("Black Lion not found:", hero)

        statement = select(Hero).where(Hero.name == "Princess Sure-E")
        result = session.exec(statement)
        hero = result.first()
        print("Princess Sure-E not found:", hero)

# Code below omitted 👇

ondelete를 SET NULL로 사용하기

관련 레코드(team 테이블의 것)가 삭제될 때 외래 키(hero 테이블의 team_id)를 NULL로 설정하도록 데이터베이스를 구성할 수도 있어요.

이 경우 Relationship() 쪽에는 cascade_delete가 없지만, Field()와 foreign_key가 있는 쪽에는 ondelete="SET NULL"이 있어요.

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


class Team(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    headquarters: str

    heroes: list["Hero"] = Relationship(back_populates="team")


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)

    team_id: int | None = Field(
        default=None, foreign_key="team.id", ondelete="SET NULL"
    )
    team: Team | None = Relationship(back_populates="heroes")

# Code below omitted 👇

위 설정은 Hero 테이블의 team_id 컬럼에 ON DELETE SET NULL이 있도록 만들어요.

이렇게 하면 누군가 SQL로 직접 데이터베이스에서 팀을 삭제할 때, 데이터베이스가 그 팀의 히어로들로 가서 team_id를 NULL로 설정해요(데이터베이스가 지원한다면요).

💡 팁

외래 키는 None 값(데이터베이스의 NULL)을 허용해야 해요. 그렇지 않으면 NOT NULL 제약을 위반해서 Integrity Error가 생기게 돼요.

그래서 team_id는 None을 포함한 타입이어야 해요. 예를 들면:

team_id: int | None

ondelete="SET NULL"을 쓰지 않으면?

ondelete="SET NULL"을 쓰지 않고, cascade_delete에도 아무것도 설정하지 않은 채로 팀을 삭제하면 어떻게 될까요?

기본 동작은 SQLModel(정확히는 SQLAlchemy)이 히어로들로 가서 Python 코드에서 그들의 team_id를 NULL로 설정하는 거예요.

그래서 기본적으로 그 team_id 필드들은 NULL로 설정돼요.

하지만 누군가 데이터베이스에 가서 직접 팀을 삭제하면, 히어로들은 존재하지 않는 팀을 가리키는 team_id를 갖게 될 수 있어요.

ondelete="SET NULL"을 추가하면 데이터베이스 자체도 그 필드들을 NULL로 설정하도록 구성돼요.

하지만 코드로 팀을 삭제하면, 기본적으로 SQLModel(SQLAlchemy)은 데이터베이스 SET NULL이 적용되기 전에도 그 team_id 필드들을 NULL로 업데이트해요.

SET NULL로 팀 삭제하기

팀을 삭제하는 코드는 이전과 똑같아요. 달라지는 건 그 아래 데이터베이스의 설정뿐이에요.

# Code above omitted 👆

def delete_team():
    with Session(engine) as session:
        statement = select(Team).where(Team.name == "Wakaland")
        team = session.exec(statement).one()
        session.delete(team)
        session.commit()
        print("Deleted team:", team)

# Code below omitted 👇

👀 전체 파일 미리보기 (Full file preview)

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


class Team(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    headquarters: str

    heroes: list["Hero"] = Relationship(back_populates="team")


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)

    team_id: int | None = Field(
        default=None, foreign_key="team.id", ondelete="SET NULL"
    )
    team: Team | None = Relationship(back_populates="heroes")


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

engine = create_engine(sqlite_url, echo=True)


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


def create_heroes():
    with Session(engine) as session:
        team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
        team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")

        hero_deadpond = Hero(
            name="Deadpond", secret_name="Dive Wilson", team=team_z_force
        )
        hero_rusty_man = Hero(
            name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers
        )
        hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
        session.add(hero_deadpond)
        session.add(hero_rusty_man)
        session.add(hero_spider_boy)
        session.commit()

        session.refresh(hero_deadpond)
        session.refresh(hero_rusty_man)
        session.refresh(hero_spider_boy)

        print("Created hero:", hero_deadpond)
        print("Created hero:", hero_rusty_man)
        print("Created hero:", hero_spider_boy)

        hero_spider_boy.team = team_preventers
        session.add(hero_spider_boy)
        session.commit()
        session.refresh(hero_spider_boy)
        print("Updated hero:", hero_spider_boy)

        hero_black_lion = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
        hero_sure_e = Hero(name="Princess Sure-E", secret_name="Sure-E")
        team_wakaland = Team(
            name="Wakaland",
            headquarters="Wakaland Capital City",
            heroes=[hero_black_lion, hero_sure_e],
        )
        session.add(team_wakaland)
        session.commit()
        session.refresh(team_wakaland)
        print("Team Wakaland:", team_wakaland)


def delete_team():
    with Session(engine) as session:
        statement = select(Team).where(Team.name == "Wakaland")
        team = session.exec(statement).one()
        session.delete(team)
        session.commit()
        print("Deleted team:", team)


def select_deleted_heroes():
    with Session(engine) as session:
        statement = select(Hero).where(Hero.name == "Black Lion")
        result = session.exec(statement)
        hero = result.first()
        print("Black Lion has no team:", hero)

        statement = select(Hero).where(Hero.name == "Princess Sure-E")
        result = session.exec(statement)
        hero = result.first()
        print("Princess Sure-E has no team:", hero)


def main():
    create_db_and_tables()
    create_heroes()
    delete_team()
    select_deleted_heroes()


if __name__ == "__main__":
    main()

이 결과로 테이블은 이렇게 남게 돼요.

SET NULL 후 Team Table

id	name	headquarters
1	Z-Force	Sister Margaret's Bar
2	Preventers	Sharp Tower

SET NULL 후 Hero Table

id	name	secret_name	age	team_id
1	Deadpond	Dive Wilson		1
2	Rusty-Man	Tommy Sharp	48	2
3	Spider-Boy	Pedro Parqueador		2
4	Black Lion	Trevor Challa	35	NULL
5	Princess Sure-E	Sure-E		NULL

Wakaland 팀은 삭제됐고, 그 히어로들은 모두 팀 없이, 다시 말해 team_id가 NULL로 설정된 채로 남았지만 데이터베이스에는 여전히 남아 있어요! 🤓

passive_deletes로 데이터베이스에 맡기기

앞선 예시들에서 우리는 ondelete를 CASCADE와 SET NULL로 설정해 데이터베이스가 관련 레코드 삭제를 자동으로 처리하도록 구성했어요. 하지만 실제로는 그 기능을 직접 써 본 적이 없어요. 왜냐하면 SQLModel(SQLAlchemy)은 기본적으로 팀의 DELETE를 보내기 전에 관련 레코드를 로드해서 삭제하거나 NULL로 업데이트하기 때문이죠.

만약 데이터베이스가 삭제/업데이트를 그 자체로 올바르게 처리할 수 있다는 걸 안다면, ondelete="CASCADE"나 ondelete="SET NULL"만으로 충분해요. 그리고 Relationship()에서 passive_deletes="all"을 사용해서 SQLModel(SQLAlchemy)에게 팀의 DELETE를 보내기 전에 그 레코드들(히어로)을 삭제하거나 업데이트하지 말라고 알려 줄 수 있어요.

SQLite에서 외래 키 지원 활성화

SQLite로 이것을 테스트하려면 먼저 외래 키 지원을 활성화해야 해요.

# Code above omitted 👆

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

# Code below omitted 👇

📌 참고

SQLite, 외래 키, 이 SQL 명령에 대해 더 자세한 내용은 SQLAlchemy 문서에서 배울 수 있어요.

passive_deletes="all" 사용하기

이제 Team의 테이블 모델을 업데이트해서 heroes의 Relationship()에 passive_deletes="all"을 사용해 볼게요.

또한 Hero 모델 테이블에서 team_id의 외래 키 Field()에 ondelete="SET NULL"을 사용해서 데이터베이스가 그 필드들을 자동으로 NULL로 설정하게 할 거예요.

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


class Team(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    headquarters: str

    heroes: list["Hero"] = Relationship(back_populates="team", passive_deletes="all")


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)

    team_id: int | None = Field(
        default=None, foreign_key="team.id", ondelete="SET NULL"
    )
    team: Team | None = Relationship(back_populates="heroes")

# Code below omitted 👇

👀 전체 파일 미리보기 (Full file preview)

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


class Team(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    headquarters: str

    heroes: list["Hero"] = Relationship(back_populates="team", passive_deletes="all")


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)

    team_id: int | None = Field(
        default=None, foreign_key="team.id", ondelete="SET NULL"
    )
    team: Team | None = Relationship(back_populates="heroes")


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

engine = create_engine(sqlite_url, echo=True)


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_heroes():
    with Session(engine) as session:
        team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
        team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")

        hero_deadpond = Hero(
            name="Deadpond", secret_name="Dive Wilson", team=team_z_force
        )
        hero_rusty_man = Hero(
            name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers
        )
        hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
        session.add(hero_deadpond)
        session.add(hero_rusty_man)
        session.add(hero_spider_boy)
        session.commit()

        session.refresh(hero_deadpond)
        session.refresh(hero_rusty_man)
        session.refresh(hero_spider_boy)

        print("Created hero:", hero_deadpond)
        print("Created hero:", hero_rusty_man)
        print("Created hero:", hero_spider_boy)

        hero_spider_boy.team = team_preventers
        session.add(hero_spider_boy)
        session.commit()
        session.refresh(hero_spider_boy)
        print("Updated hero:", hero_spider_boy)

        hero_black_lion = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
        hero_sure_e = Hero(name="Princess Sure-E", secret_name="Sure-E")
        team_wakaland = Team(
            name="Wakaland",
            headquarters="Wakaland Capital City",
            heroes=[hero_black_lion, hero_sure_e],
        )
        session.add(team_wakaland)
        session.commit()
        session.refresh(team_wakaland)
        print("Team Wakaland:", team_wakaland)


def delete_team():
    with Session(engine) as session:
        statement = select(Team).where(Team.name == "Wakaland")
        team = session.exec(statement).one()
        session.delete(team)
        session.commit()
        print("Deleted team:", team)


def select_deleted_heroes():
    with Session(engine) as session:
        statement = select(Hero).where(Hero.name == "Black Lion")
        result = session.exec(statement)
        hero = result.first()
        print("Black Lion has no team:", hero)

        statement = select(Hero).where(Hero.name == "Princess Sure-E")
        result = session.exec(statement)
        hero = result.first()
        print("Princess Sure-E has no team:", hero)


def main():
    create_db_and_tables()
    create_heroes()
    delete_team()
    select_deleted_heroes()


if __name__ == "__main__":
    main()

ondelete를 RESTRICT로 사용하기

관련 레코드(히어로)가 있을 때 레코드(팀)의 삭제를 막도록 데이터베이스를 구성할 수도 있어요.

이 경우 누군가 히어로가 있는 팀을 삭제하려 하면 데이터베이스가 오류를 일으켜요.

그리고 이것은 데이터베이스에 구성되므로, 누군가 SQL을 사용해서 데이터베이스에 직접 접근해도 (데이터베이스가 지원한다면) 동일하게 동작해요.

💡 팁

SQLite에서는 이것도 외래 키 지원을 활성화해야 해요.

RESTRICT용으로 SQLite 외래 키 지원 활성화

ondelete="RESTRICT"는 주로 데이터베이스 수준의 제약이므로, 테스트할 수 있도록 먼저 SQLite에서 외래 키 지원을 활성화해 볼게요.

# Code above omitted 👆

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

# Code below omitted 👇

ondelete="RESTRICT" 사용하기

Hero 모델 테이블에서 team_id의 외래 키 Field()에 ondelete="RESTRICT"를 설정해 볼게요.

그리고 Team 모델 테이블에서는 heroes의 Relationship()에 passive_deletes="all"을 사용할 거예요. 이렇게 하면 삭제된 모델의 외래 키를 NULL로 설정하는 기본 동작이 비활성화되고, 히어로가 있는 팀을 삭제하려 하면 데이터베이스가 오류를 일으킬 거예요.

💡 팁

우리가 Team 모델 테이블에는 cascade_delete를 설정하지 않는다는 점을 눈여겨보세요.

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


class Team(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    headquarters: str

    heroes: list["Hero"] = Relationship(back_populates="team", passive_deletes="all")


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)

    team_id: int | None = Field(
        default=None, foreign_key="team.id", ondelete="RESTRICT"
    )
    team: Team | None = Relationship(back_populates="heroes")

# Code below omitted 👇

👀 전체 파일 미리보기 (Full file preview)

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


class Team(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    headquarters: str

    heroes: list["Hero"] = Relationship(back_populates="team", passive_deletes="all")


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)

    team_id: int | None = Field(
        default=None, foreign_key="team.id", ondelete="RESTRICT"
    )
    team: Team | None = Relationship(back_populates="heroes")


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

engine = create_engine(sqlite_url, echo=True)


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_heroes():
    with Session(engine) as session:
        team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
        team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")

        hero_deadpond = Hero(
            name="Deadpond", secret_name="Dive Wilson", team=team_z_force
        )
        hero_rusty_man = Hero(
            name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers
        )
        hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
        session.add(hero_deadpond)
        session.add(hero_rusty_man)
        session.add(hero_spider_boy)
        session.commit()

        session.refresh(hero_deadpond)
        session.refresh(hero_rusty_man)
        session.refresh(hero_spider_boy)

        print("Created hero:", hero_deadpond)
        print("Created hero:", hero_rusty_man)
        print("Created hero:", hero_spider_boy)

        hero_spider_boy.team = team_preventers
        session.add(hero_spider_boy)
        session.commit()
        session.refresh(hero_spider_boy)
        print("Updated hero:", hero_spider_boy)

        hero_black_lion = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
        hero_sure_e = Hero(name="Princess Sure-E", secret_name="Sure-E")
        team_wakaland = Team(
            name="Wakaland",
            headquarters="Wakaland Capital City",
            heroes=[hero_black_lion, hero_sure_e],
        )
        session.add(team_wakaland)
        session.commit()
        session.refresh(team_wakaland)
        print("Team Wakaland:", team_wakaland)


def delete_team():
    with Session(engine) as session:
        statement = select(Team).where(Team.name == "Wakaland")
        team = session.exec(statement).one()
        session.delete(team)
        session.commit()
        print("Deleted team:", team)


def select_deleted_heroes():
    with Session(engine) as session:
        statement = select(Hero).where(Hero.name == "Black Lion")
        result = session.exec(statement)
        hero = result.first()
        print("Black Lion has no team:", hero)

        statement = select(Hero).where(Hero.name == "Princess Sure-E")
        result = session.exec(statement)
        hero = result.first()
        print("Princess Sure-E has no team:", hero)


def main():
    create_db_and_tables()
    create_heroes()
    delete_team()


if __name__ == "__main__":
    main()

프로그램 실행, 오류 확인하기

이제 프로그램을 실행해서 히어로가 있는 팀을 삭제하려 하면 오류가 나는 걸 보게 될 거예요.

좋아요! 데이터베이스가 히어로가 있는 팀을 삭제하는 실수를 커밋하지 못하게 막아 줬어요. 🤓

💡 팁

PRAGMA foreign_keys=ON이 필요한지 테스트해 보려면 그 줄을 주석 처리하고 다시 실행해 보세요. 오류가 나지 않을 거예요. 😱

passive_deletes="all"도 마찬가지예요. 그 줄을 주석 처리하면 SQLModel(SQLAlchemy)이 팀을 삭제하기 전에 히어로들을 로드해서 그들의 외래 키 team_id를 NULL로 설정하므로, 제약이 기대처럼 동작하지 않아 오류가 나지 않을 거예요. 😅

팀을 삭제하기 전에 히어로들 업데이트하기

ondelete="RESTRICT"를 설정하고, SQLite가 외래 키를 지원하도록 구성하고, Relationship()에 passive_deletes="all"을 둔 상태에서, 히어로가 있는 팀을 삭제하려 하면 오류가 나요.

팀을 삭제하려면 먼저 히어로들을 업데이트해서 그들의 team_id를 None(데이터베이스의 NULL)으로 설정해야 해요.

리스트의 .clear() 메서드를 호출하면 모든 항목이 제거되죠. 그래서 team.heroes.clear()를 호출하고 그걸 데이터베이스에 저장하면, 히어로들을 팀에서 분리하고 그들의 team_id가 None으로 설정돼요.

💡 팁

team.heroes.clear()를 호출하는 것은 우리가 passive_deletes="all"을 설정하지 않았다면 SQLModel(SQLAlchemy)이 했을 것과 매우 비슷해요.

# Code above omitted 👆

def remove_team_heroes():
    with Session(engine) as session:
        statement = select(Team).where(Team.name == "Wakaland")
        team = session.exec(statement).one()
        team.heroes.clear()
        session.add(team)
        session.commit()
        session.refresh(team)
        print("Team with removed heroes:", team)

# Code below omitted 👇

👀 전체 파일 미리보기 (Full file preview)

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


class Team(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    headquarters: str

    heroes: list["Hero"] = Relationship(back_populates="team", passive_deletes="all")


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)

    team_id: int | None = Field(
        default=None, foreign_key="team.id", ondelete="RESTRICT"
    )
    team: Team | None = Relationship(back_populates="heroes")


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

engine = create_engine(sqlite_url, echo=True)


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_heroes():
    with Session(engine) as session:
        team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
        team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")

        hero_deadpond = Hero(
            name="Deadpond", secret_name="Dive Wilson", team=team_z_force
        )
        hero_rusty_man = Hero(
            name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers
        )
        hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
        session.add(hero_deadpond)
        session.add(hero_rusty_man)
        session.add(hero_spider_boy)
        session.commit()

        session.refresh(hero_deadpond)
        session.refresh(hero_rusty_man)
        session.refresh(hero_spider_boy)

        print("Created hero:", hero_deadpond)
        print("Created hero:", hero_rusty_man)
        print("Created hero:", hero_spider_boy)

        hero_spider_boy.team = team_preventers
        session.add(hero_spider_boy)
        session.commit()
        session.refresh(hero_spider_boy)
        print("Updated hero:", hero_spider_boy)

        hero_black_lion = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
        hero_sure_e = Hero(name="Princess Sure-E", secret_name="Sure-E")
        team_wakaland = Team(
            name="Wakaland",
            headquarters="Wakaland Capital City",
            heroes=[hero_black_lion, hero_sure_e],
        )
        session.add(team_wakaland)
        session.commit()
        session.refresh(team_wakaland)
        print("Team Wakaland:", team_wakaland)


def remove_team_heroes():
    with Session(engine) as session:
        statement = select(Team).where(Team.name == "Wakaland")
        team = session.exec(statement).one()
        team.heroes.clear()
        session.add(team)
        session.commit()
        session.refresh(team)
        print("Team with removed heroes:", team)


def delete_team():
    with Session(engine) as session:
        statement = select(Team).where(Team.name == "Wakaland")
        team = session.exec(statement).one()
        session.delete(team)
        session.commit()
        print("Deleted team:", team)


def select_deleted_heroes():
    with Session(engine) as session:
        statement = select(Hero).where(Hero.name == "Black Lion")
        result = session.exec(statement)
        hero = result.first()
        print("Black Lion has no team:", hero)

        statement = select(Hero).where(Hero.name == "Princess Sure-E")
        result = session.exec(statement)
        hero = result.first()
        print("Princess Sure-E has no team:", hero)


def main():
    create_db_and_tables()
    create_heroes()
    remove_team_heroes()
    delete_team()
    select_deleted_heroes()


if __name__ == "__main__":
    main()

결론

많은 경우에 사실 아무것도 설정할 필요가 없어요. 😎

어떤 경우에는 레코드의 삭제를 관련 레코드들로 자동 연쇄(팀을 그 히어로들과 함께 삭제)하고 싶을 때, 이렇게 하면 돼요:

외래 키가 없는 쪽의 Relationship()에서 cascade_delete=True 사용
그리고 외래 키가 있는 Field()에서 ondelete="CASCADE" 사용

이걸로 대부분의 사용 사례를 해결할 수 있어요. 🚀

그리고 다른 게 필요하다면, 위에서 설명한 추가 옵션들을 참고하면 돼요. 🤓

관계 삭제 연쇄하기