mirror of
https://github.com/fastapi/sqlmodel.git
synced 2025-08-15 02:07:54 +08:00
✨ Add support for cascade delete relationships: cascade_delete, ondelete, and passive_deletes (#983)
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: Sebastián Ramírez <tiangolo@gmail.com>
This commit is contained in:
Esteban Maya
@ -0,0 +1,110 @@
|
||||
from typing import List, Optional
|
||||
|
||||
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select
|
||||
|
||||
|
||||
class Team(SQLModel, table=True):
|
||||
id: Optional[int] = 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: Optional[int] = Field(default=None, primary_key=True)
|
||||
name: str = Field(index=True)
|
||||
secret_name: str
|
||||
age: Optional[int] = Field(default=None, index=True)
|
||||
|
||||
team_id: Optional[int] = Field(
|
||||
default=None, foreign_key="team.id", ondelete="CASCADE"
|
||||
)
|
||||
team: Optional[Team] = 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()
|
||||
@ -0,0 +1,106 @@
|
||||
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()
|
||||
@ -0,0 +1,110 @@
|
||||
from typing import Optional
|
||||
|
||||
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select
|
||||
|
||||
|
||||
class Team(SQLModel, table=True):
|
||||
id: Optional[int] = 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: Optional[int] = Field(default=None, primary_key=True)
|
||||
name: str = Field(index=True)
|
||||
secret_name: str
|
||||
age: Optional[int] = Field(default=None, index=True)
|
||||
|
||||
team_id: Optional[int] = Field(
|
||||
default=None, foreign_key="team.id", ondelete="CASCADE"
|
||||
)
|
||||
team: Optional[Team] = 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()
|
||||
@ -0,0 +1,110 @@
|
||||
from typing import List, Optional
|
||||
|
||||
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select
|
||||
|
||||
|
||||
class Team(SQLModel, table=True):
|
||||
id: Optional[int] = 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: Optional[int] = Field(default=None, primary_key=True)
|
||||
name: str = Field(index=True)
|
||||
secret_name: str
|
||||
age: Optional[int] = Field(default=None, index=True)
|
||||
|
||||
team_id: Optional[int] = Field(
|
||||
default=None, foreign_key="team.id", ondelete="SET NULL"
|
||||
)
|
||||
team: Optional[Team] = 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()
|
||||
@ -0,0 +1,108 @@
|
||||
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()
|
||||
@ -0,0 +1,110 @@
|
||||
from typing import Optional
|
||||
|
||||
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select
|
||||
|
||||
|
||||
class Team(SQLModel, table=True):
|
||||
id: Optional[int] = 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: Optional[int] = Field(default=None, primary_key=True)
|
||||
name: str = Field(index=True)
|
||||
secret_name: str
|
||||
age: Optional[int] = Field(default=None, index=True)
|
||||
|
||||
team_id: Optional[int] = Field(
|
||||
default=None, foreign_key="team.id", ondelete="SET NULL"
|
||||
)
|
||||
team: Optional[Team] = 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()
|
||||
@ -0,0 +1,112 @@
|
||||
from typing import List, Optional
|
||||
|
||||
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select, text
|
||||
|
||||
|
||||
class Team(SQLModel, table=True):
|
||||
id: Optional[int] = 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: Optional[int] = Field(default=None, primary_key=True)
|
||||
name: str = Field(index=True)
|
||||
secret_name: str
|
||||
age: Optional[int] = Field(default=None, index=True)
|
||||
|
||||
team_id: Optional[int] = Field(
|
||||
default=None, foreign_key="team.id", ondelete="SET NULL"
|
||||
)
|
||||
team: Optional[Team] = 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()
|
||||
@ -0,0 +1,110 @@
|
||||
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()
|
||||
@ -0,0 +1,112 @@
|
||||
from typing import Optional
|
||||
|
||||
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select, text
|
||||
|
||||
|
||||
class Team(SQLModel, table=True):
|
||||
id: Optional[int] = 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: Optional[int] = Field(default=None, primary_key=True)
|
||||
name: str = Field(index=True)
|
||||
secret_name: str
|
||||
age: Optional[int] = Field(default=None, index=True)
|
||||
|
||||
team_id: Optional[int] = Field(
|
||||
default=None, foreign_key="team.id", ondelete="SET NULL"
|
||||
)
|
||||
team: Optional[Team] = 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()
|
||||
@ -0,0 +1,111 @@
|
||||
from typing import List, Optional
|
||||
|
||||
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select, text
|
||||
|
||||
|
||||
class Team(SQLModel, table=True):
|
||||
id: Optional[int] = 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: Optional[int] = Field(default=None, primary_key=True)
|
||||
name: str = Field(index=True)
|
||||
secret_name: str
|
||||
age: Optional[int] = Field(default=None, index=True)
|
||||
|
||||
team_id: Optional[int] = Field(
|
||||
default=None, foreign_key="team.id", ondelete="RESTRICT"
|
||||
)
|
||||
team: Optional[Team] = 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()
|
||||
@ -0,0 +1,109 @@
|
||||
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()
|
||||
@ -0,0 +1,111 @@
|
||||
from typing import Optional
|
||||
|
||||
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select, text
|
||||
|
||||
|
||||
class Team(SQLModel, table=True):
|
||||
id: Optional[int] = 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: Optional[int] = Field(default=None, primary_key=True)
|
||||
name: str = Field(index=True)
|
||||
secret_name: str
|
||||
age: Optional[int] = Field(default=None, index=True)
|
||||
|
||||
team_id: Optional[int] = Field(
|
||||
default=None, foreign_key="team.id", ondelete="RESTRICT"
|
||||
)
|
||||
team: Optional[Team] = 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()
|
||||
@ -0,0 +1,124 @@
|
||||
from typing import List, Optional
|
||||
|
||||
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select, text
|
||||
|
||||
|
||||
class Team(SQLModel, table=True):
|
||||
id: Optional[int] = 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: Optional[int] = Field(default=None, primary_key=True)
|
||||
name: str = Field(index=True)
|
||||
secret_name: str
|
||||
age: Optional[int] = Field(default=None, index=True)
|
||||
|
||||
team_id: Optional[int] = Field(
|
||||
default=None, foreign_key="team.id", ondelete="RESTRICT"
|
||||
)
|
||||
team: Optional[Team] = 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()
|
||||
@ -0,0 +1,122 @@
|
||||
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()
|
||||
@ -0,0 +1,124 @@
|
||||
from typing import Optional
|
||||
|
||||
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select, text
|
||||
|
||||
|
||||
class Team(SQLModel, table=True):
|
||||
id: Optional[int] = 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: Optional[int] = Field(default=None, primary_key=True)
|
||||
name: str = Field(index=True)
|
||||
secret_name: str
|
||||
age: Optional[int] = Field(default=None, index=True)
|
||||
|
||||
team_id: Optional[int] = Field(
|
||||
default=None, foreign_key="team.id", ondelete="RESTRICT"
|
||||
)
|
||||
team: Optional[Team] = 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()
|
||||
Reference in New Issue
Block a user