Aliased Relationships¶
Multiple Relationships to the Same Model¶
We've seen how tables are related to each other via a single relationship attribute but what if more than one attribute links to the same table?
What if you have a User
model and an Address
model and would like
to have User.home_address
and User.work_address
relationships to the same
Address
model? In SQL you do this by creating a table alias using AS
like this:
SELECT *
FROM user
JOIN address AS home_address_alias
ON user.home_address_id == home_address_alias.id
JOIN address AS work_address_alias
ON user.work_address_id == work_address_alias.id
The aliases we create are home_address_alias
and work_address_alias
. You can think of them
as a view to the same underlying address
table.
We can do this with SQLModel and SQLAlchemy using sqlalchemy.orm.aliased
and a couple of extra bits of info in our SQLModel relationship definition and join statements.
The Relationships¶
Let's define a winter_team
and summer_team
relationship for our heros. They can be on different
winter and summer teams or on the same team for both seasons.
# Code above omitted 👆
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)
winter_team_id: Optional[int] = Field(default=None, foreign_key="team.id")
winter_team: Optional[Team] = Relationship(
sa_relationship_kwargs={"primaryjoin": "Hero.winter_team_id == Team.id"}
)
summer_team_id: Optional[int] = Field(default=None, foreign_key="team.id")
summer_team: Optional[Team] = Relationship(
sa_relationship_kwargs={"primaryjoin": "Hero.summer_team_id == Team.id"}
)
# Code below omitted 👇
👀 Full file preview
from typing import Optional
from sqlalchemy.orm import aliased
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
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)
winter_team_id: Optional[int] = Field(default=None, foreign_key="team.id")
winter_team: Optional[Team] = Relationship(
sa_relationship_kwargs={"primaryjoin": "Hero.winter_team_id == Team.id"}
)
summer_team_id: Optional[int] = Field(default=None, foreign_key="team.id")
summer_team: Optional[Team] = Relationship(
sa_relationship_kwargs={"primaryjoin": "Hero.summer_team_id == Team.id"}
)
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",
winter_team=team_preventers,
summer_team=team_z_force,
)
hero_rusty_man = Hero(
name="Rusty-Man",
secret_name="Tommy Sharp",
age=48,
winter_team=team_preventers,
summer_team=team_preventers,
)
session.add(hero_deadpond)
session.add(hero_rusty_man)
session.commit()
session.refresh(hero_deadpond)
session.refresh(hero_rusty_man)
print("Created hero:", hero_deadpond)
print("Created hero:", hero_rusty_man)
def select_heroes():
with Session(engine) as session:
winter_alias = aliased(Team)
# Heros with winter team as the Preventers
result = session.exec(
select(Hero)
.join(winter_alias, onclause=Hero.winter_team_id == winter_alias.id)
.where(winter_alias.name == "Preventers")
)
heros = result.all()
print("Heros with Preventers as their winter team:", heros)
assert len(heros) == 2
summer_alias = aliased(Team)
# Heros with Preventers as their winter team and Z-Force as their summer team
result = session.exec(
select(Hero)
.join(winter_alias, onclause=Hero.winter_team_id == winter_alias.id)
.where(winter_alias.name == "Preventers")
.join(summer_alias, onclause=Hero.summer_team_id == summer_alias.id)
.where(summer_alias.name == "Z-Force")
)
heros = result.all()
print(
"Heros with Preventers as their winter and Z-Force as their summer team:",
heros,
)
assert len(heros) == 1
assert heros[0].name == "Deadpond"
def main():
create_db_and_tables()
create_heroes()
select_heroes()
if __name__ == "__main__":
main()
The sa_relationship_kwargs={"primaryjoin": ...}
is a new bit of info we need for SQLAlchemy to
figure out which SQL join we should use depending on which attribute is in our query.
Creating Heros¶
Creating Heros
with the multiple teams is no different from before. We set the same or different
team to the winter_team
and summer_team
attributes:
# Code above omitted 👆
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",
winter_team=team_preventers,
summer_team=team_z_force,
)
hero_rusty_man = Hero(
name="Rusty-Man",
secret_name="Tommy Sharp",
age=48,
winter_team=team_preventers,
summer_team=team_preventers,
)
session.add(hero_deadpond)
session.add(hero_rusty_man)
session.commit()
session.refresh(hero_deadpond)
session.refresh(hero_rusty_man)
print("Created hero:", hero_deadpond)
print("Created hero:", hero_rusty_man)
# Code below omitted 👇
👀 Full file preview
from typing import Optional
from sqlalchemy.orm import aliased
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
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)
winter_team_id: Optional[int] = Field(default=None, foreign_key="team.id")
winter_team: Optional[Team] = Relationship(
sa_relationship_kwargs={"primaryjoin": "Hero.winter_team_id == Team.id"}
)
summer_team_id: Optional[int] = Field(default=None, foreign_key="team.id")
summer_team: Optional[Team] = Relationship(
sa_relationship_kwargs={"primaryjoin": "Hero.summer_team_id == Team.id"}
)
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",
winter_team=team_preventers,
summer_team=team_z_force,
)
hero_rusty_man = Hero(
name="Rusty-Man",
secret_name="Tommy Sharp",
age=48,
winter_team=team_preventers,
summer_team=team_preventers,
)
session.add(hero_deadpond)
session.add(hero_rusty_man)
session.commit()
session.refresh(hero_deadpond)
session.refresh(hero_rusty_man)
print("Created hero:", hero_deadpond)
print("Created hero:", hero_rusty_man)
def select_heroes():
with Session(engine) as session:
winter_alias = aliased(Team)
# Heros with winter team as the Preventers
result = session.exec(
select(Hero)
.join(winter_alias, onclause=Hero.winter_team_id == winter_alias.id)
.where(winter_alias.name == "Preventers")
)
heros = result.all()
print("Heros with Preventers as their winter team:", heros)
assert len(heros) == 2
summer_alias = aliased(Team)
# Heros with Preventers as their winter team and Z-Force as their summer team
result = session.exec(
select(Hero)
.join(winter_alias, onclause=Hero.winter_team_id == winter_alias.id)
.where(winter_alias.name == "Preventers")
.join(summer_alias, onclause=Hero.summer_team_id == summer_alias.id)
.where(summer_alias.name == "Z-Force")
)
heros = result.all()
print(
"Heros with Preventers as their winter and Z-Force as their summer team:",
heros,
)
assert len(heros) == 1
assert heros[0].name == "Deadpond"
def main():
create_db_and_tables()
create_heroes()
select_heroes()
if __name__ == "__main__":
main()
Searching for Heros¶
Querying Heros
based on the winter or summer teams adds a bit of complication. We need to create the
alias and we also need to be a bit more explicit in how we tell SQLAlchemy to join the hero
and team
tables.
We create the alias using sqlalchemy.orm.aliased
function and use the alias in the where
function. We also
need to provide an onclause
argument to the join
.
# Code above omitted 👆
winter_alias = aliased(Team)
# Heros with winter team as the Preventers
result = session.exec(
select(Hero)
.join(winter_alias, onclause=Hero.winter_team_id == winter_alias.id)
.where(winter_alias.name == "Preventers")
)
heros = result.all()
print("Heros with Preventers as their winter team:", heros)
# Code below omitted 👇
👀 Full file preview
from typing import Optional
from sqlalchemy.orm import aliased
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
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)
winter_team_id: Optional[int] = Field(default=None, foreign_key="team.id")
winter_team: Optional[Team] = Relationship(
sa_relationship_kwargs={"primaryjoin": "Hero.winter_team_id == Team.id"}
)
summer_team_id: Optional[int] = Field(default=None, foreign_key="team.id")
summer_team: Optional[Team] = Relationship(
sa_relationship_kwargs={"primaryjoin": "Hero.summer_team_id == Team.id"}
)
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",
winter_team=team_preventers,
summer_team=team_z_force,
)
hero_rusty_man = Hero(
name="Rusty-Man",
secret_name="Tommy Sharp",
age=48,
winter_team=team_preventers,
summer_team=team_preventers,
)
session.add(hero_deadpond)
session.add(hero_rusty_man)
session.commit()
session.refresh(hero_deadpond)
session.refresh(hero_rusty_man)
print("Created hero:", hero_deadpond)
print("Created hero:", hero_rusty_man)
def select_heroes():
with Session(engine) as session:
winter_alias = aliased(Team)
# Heros with winter team as the Preventers
result = session.exec(
select(Hero)
.join(winter_alias, onclause=Hero.winter_team_id == winter_alias.id)
.where(winter_alias.name == "Preventers")
)
heros = result.all()
print("Heros with Preventers as their winter team:", heros)
assert len(heros) == 2
summer_alias = aliased(Team)
# Heros with Preventers as their winter team and Z-Force as their summer team
result = session.exec(
select(Hero)
.join(winter_alias, onclause=Hero.winter_team_id == winter_alias.id)
.where(winter_alias.name == "Preventers")
.join(summer_alias, onclause=Hero.summer_team_id == summer_alias.id)
.where(summer_alias.name == "Z-Force")
)
heros = result.all()
print(
"Heros with Preventers as their winter and Z-Force as their summer team:",
heros,
)
assert len(heros) == 1
assert heros[0].name == "Deadpond"
def main():
create_db_and_tables()
create_heroes()
select_heroes()
if __name__ == "__main__":
main()
The value for the onclause
is the same value that you used in the primaryjoin
argument
when the relationship is defined in the Hero
model.
To use both team attributes in a query, create another alias
and add the join:
# Code above omitted 👆
summer_alias = aliased(Team)
# Heros with Preventers as their winter team and Z-Force as their summer team
result = session.exec(
select(Hero)
.join(winter_alias, onclause=Hero.winter_team_id == winter_alias.id)
.where(winter_alias.name == "Preventers")
.join(summer_alias, onclause=Hero.summer_team_id == summer_alias.id)
.where(summer_alias.name == "Z-Force")
)
heros = result.all()
print(
"Heros with Preventers as their winter and Z-Force as their summer team:",
heros,
)
# Code below omitted 👇
👀 Full file preview
from typing import Optional
from sqlalchemy.orm import aliased
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
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)
winter_team_id: Optional[int] = Field(default=None, foreign_key="team.id")
winter_team: Optional[Team] = Relationship(
sa_relationship_kwargs={"primaryjoin": "Hero.winter_team_id == Team.id"}
)
summer_team_id: Optional[int] = Field(default=None, foreign_key="team.id")
summer_team: Optional[Team] = Relationship(
sa_relationship_kwargs={"primaryjoin": "Hero.summer_team_id == Team.id"}
)
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",
winter_team=team_preventers,
summer_team=team_z_force,
)
hero_rusty_man = Hero(
name="Rusty-Man",
secret_name="Tommy Sharp",
age=48,
winter_team=team_preventers,
summer_team=team_preventers,
)
session.add(hero_deadpond)
session.add(hero_rusty_man)
session.commit()
session.refresh(hero_deadpond)
session.refresh(hero_rusty_man)
print("Created hero:", hero_deadpond)
print("Created hero:", hero_rusty_man)
def select_heroes():
with Session(engine) as session:
winter_alias = aliased(Team)
# Heros with winter team as the Preventers
result = session.exec(
select(Hero)
.join(winter_alias, onclause=Hero.winter_team_id == winter_alias.id)
.where(winter_alias.name == "Preventers")
)
heros = result.all()
print("Heros with Preventers as their winter team:", heros)
assert len(heros) == 2
summer_alias = aliased(Team)
# Heros with Preventers as their winter team and Z-Force as their summer team
result = session.exec(
select(Hero)
.join(winter_alias, onclause=Hero.winter_team_id == winter_alias.id)
.where(winter_alias.name == "Preventers")
.join(summer_alias, onclause=Hero.summer_team_id == summer_alias.id)
.where(summer_alias.name == "Z-Force")
)
heros = result.all()
print(
"Heros with Preventers as their winter and Z-Force as their summer team:",
heros,
)
assert len(heros) == 1
assert heros[0].name == "Deadpond"
def main():
create_db_and_tables()
create_heroes()
select_heroes()
if __name__ == "__main__":
main()