SQLModel 关联查询

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


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


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")


def test():
    sqlite_file_name = "database.db"
    sqlite_url = f"sqlite:///{sqlite_file_name}"
    engine = create_engine(sqlite_url, echo=True)
    SQLModel.metadata.drop_all(engine)
    SQLModel.metadata.create_all(engine)

    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")
        team_empty = Team(name="Empty", headquarters="Empty Team")
        session.add(team_preventers)
        session.add(team_z_force)
        session.add(team_empty)
        session.commit()

        hero_deadpond = Hero(
            name="Deadpond", secret_name="Dive Wilson", team_id=team_z_force.id,
        )
        hero_rusty_man = Hero(
            name="Rusty-Man",
            secret_name="Tommy Sharp",
            age=48,
            team_id=team_preventers.id,
        )
        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()

    print("笛卡尔积+条件筛选")
    print("#"*80)
    with Session(engine) as session:
        statement = select(Hero, Team).where(Hero.team_id == Team.id)
        results = session.exec(statement)
        for hero, team in results:
            print("Hero:", hero, "Team:", team)

    print("Inner Join")
    print("#"*80)
    with Session(engine) as session:
        statement = select(Hero, Team).join(Team)
        results = session.exec(statement)
        for hero, team in results:
            print("Hero:", hero, "Team:", team)

    print("Left Outer Join")
    print("#"*80)
    with Session(engine) as session:
        statement = select(Hero, Team).join(Team, isouter=True)
        results = session.exec(statement)
        for hero, team in results:
            print("Hero:", hero, "Team:", team)

    # 系统不直接支持RIGHT OUTER JOIN，使用的时候把表顺序换一下就行
    print("Right Outer Join")
    print("#"*80)
    with Session(engine) as session:
        statement = select(Hero, Team).join(Hero, isouter=True)
        results = session.exec(statement)
        for hero, team in results:
            print("Hero:", hero, "Team:", team)

    # 数据库不直接支持FULL OUTER JOIN，使用UNION进行结果集合并。
    # 两个数据集合并后丢失数据模型信息，不能直接用for hero, team in results
    print("Full Outer Join")
    with Session(engine) as session:
        statement = union(
            select(Hero, Team).join(Team, isouter=True),
            select(Hero, Team).join(Hero, isouter=True),
        )
        results = session.exec(statement)
        for res in results:
            print(res)

if __name__ == "__main__":
    test()