Self-referential relationships¶
Oftentimes we need to model a relationship between one entity of some class and another entity (or multiple entities) of that same class. This is known as a self-referential or recursive relationship, sometimes also called an adjacency list.
In database terms this means having a table with a foreign key reference to the primary key in the same table.
Say, for example, we want to introduce a Villain class. 😈 Every villain can have a boss, who also must be a villain. If a villain is the boss of other villains, we want to call those his minions.
Let's implement this with SQLModel. 🤓
Using SQLAlchemy arguments¶
We already learned a lot about Relationship attributes in previous chapters. We know that SQLModel is built on top of SQLAlchemy, which supports defining self-referential relationships (see their documentation).
To allow more fine-grained control over it, the Relationship constructor allows explicitly passing additional keyword-arguments to the sqlalchemy.orm.relationship constructor that is being called under the hood via the sa_relationship_kwargs parameter. This should be a mapping (e.g. a dictionary) of strings representing the SQLAlchemy parameter names to the values we want to pass through as arguments.
Since SQLAlchemy relationships provide the remote_side parameter for just such an occasion, we can leverage that directly to construct the self-referential pattern with minimal code.
# Code above omitted 👆
class Villain(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
power_level: int
boss_id: Optional[int] = Field(
foreign_key="villain.id", default=None, nullable=True
)
boss: Optional["Villain"] = Relationship(
back_populates="minions",
sa_relationship_kwargs={"remote_side": "Villain.id"},
)
minions: List["Villain"] = Relationship(back_populates="boss")
# Code below omitted 👇
👀 Full file preview
from typing import List, Optional
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine
class Villain(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
power_level: int
boss_id: Optional[int] = Field(
foreign_key="villain.id", default=None, nullable=True
)
boss: Optional["Villain"] = Relationship(
back_populates="minions",
sa_relationship_kwargs={"remote_side": "Villain.id"},
)
minions: List["Villain"] = Relationship(back_populates="boss")
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=False)
def create_db_and_tables() -> None:
SQLModel.metadata.create_all(engine)
def create_villains() -> None:
with Session(engine) as session:
thinnus = Villain(name="Thinnus", power_level=9001)
ebonite_mew = Villain(name="Ebonite Mew", power_level=400, boss=thinnus)
dark_shorty = Villain(name="Dark Shorty", power_level=200, boss=thinnus)
ultra_bot = Villain(name="Ultra Bot", power_level=2**9)
session.add(ebonite_mew)
session.add(dark_shorty)
session.add(ultra_bot)
session.commit()
session.refresh(thinnus)
session.refresh(ebonite_mew)
session.refresh(dark_shorty)
session.refresh(ultra_bot)
print("Created villain:", thinnus)
print("Created villain:", ebonite_mew)
print("Created villain:", dark_shorty)
print("Created villain:", ultra_bot)
ultra_bot.boss = thinnus
session.add(ultra_bot)
session.commit()
session.refresh(ultra_bot)
print("Updated villain:", ultra_bot)
clone_bot_1 = Villain(name="Clone Bot 1", power_level=2**6)
clone_bot_2 = Villain(name="Clone Bot 2", power_level=2**6)
clone_bot_3 = Villain(name="Clone Bot 3", power_level=2**6)
ultra_bot.minions.extend([clone_bot_1, clone_bot_2, clone_bot_3])
session.add(ultra_bot)
session.commit()
session.refresh(clone_bot_1)
session.refresh(clone_bot_2)
session.refresh(clone_bot_3)
print("Added minion:", clone_bot_1)
print("Added minion:", clone_bot_2)
print("Added minion:", clone_bot_3)
def main() -> None:
create_db_and_tables()
create_villains()
if __name__ == "__main__":
main()
Using the sa_relationship_kwargs parameter, we pass the keyword argument remote_side='Villain.id' to the underlying relationship property.
Info
The remote_side parameter accepts a Python-evaluable string when using Declarative. This allows us to reference Villain.id even though the class is still being defined.
Alternatively, you can use a callable:
sa_relationship_kwargs={"remote_side": lambda : Villain.id}
Back-populating and self-referencing¶
Notice that we explicitly defined the relationship attributes we wanted for referring to the boss as well as the minions of a Villain.
For our purposes, it is necessary that we also provide the back_populates parameter to both relationships as explained in detail in a dedicated chapter.
In addition, the type annotations were made by enclosing our Villain class name in quotes, since we are referencing a class that is not yet fully defined by the time the interpreter reaches those lines. See the chapter on type annotation strings for a detailed explanation.
Finally, as with regular (i.e. non-self-referential) foreign key relationships, it is up to us to decide whether it makes sense to allow the field to be empty or not. In our example, not every villain must have a boss (in fact, we would otherwise introduce a circular reference chain, which would not make sense in this context). Therefore we declare boss_id: Optional[int] and boss: Optional['Villain']. This is analogous to the Hero→Team relationship we saw in an earlier chapter.
Creating instances¶
Now let's see how we can create villains with a boss:
# Code above omitted 👆
def create_villains() -> None:
with Session(engine) as session:
thinnus = Villain(name="Thinnus", power_level=9001)
ebonite_mew = Villain(name="Ebonite Mew", power_level=400, boss=thinnus)
dark_shorty = Villain(name="Dark Shorty", power_level=200, boss=thinnus)
ultra_bot = Villain(name="Ultra Bot", power_level=2**9)
session.add(ebonite_mew)
session.add(dark_shorty)
session.add(ultra_bot)
session.commit()
session.refresh(thinnus)
session.refresh(ebonite_mew)
session.refresh(dark_shorty)
session.refresh(ultra_bot)
print("Created villain:", thinnus)
print("Created villain:", ebonite_mew)
print("Created villain:", dark_shorty)
print("Created villain:", ultra_bot)
# Code below omitted 👇
👀 Full file preview
from typing import List, Optional
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine
class Villain(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
power_level: int
boss_id: Optional[int] = Field(
foreign_key="villain.id", default=None, nullable=True
)
boss: Optional["Villain"] = Relationship(
back_populates="minions",
sa_relationship_kwargs={"remote_side": "Villain.id"},
)
minions: List["Villain"] = Relationship(back_populates="boss")
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=False)
def create_db_and_tables() -> None:
SQLModel.metadata.create_all(engine)
def create_villains() -> None:
with Session(engine) as session:
thinnus = Villain(name="Thinnus", power_level=9001)
ebonite_mew = Villain(name="Ebonite Mew", power_level=400, boss=thinnus)
dark_shorty = Villain(name="Dark Shorty", power_level=200, boss=thinnus)
ultra_bot = Villain(name="Ultra Bot", power_level=2**9)
session.add(ebonite_mew)
session.add(dark_shorty)
session.add(ultra_bot)
session.commit()
session.refresh(thinnus)
session.refresh(ebonite_mew)
session.refresh(dark_shorty)
session.refresh(ultra_bot)
print("Created villain:", thinnus)
print("Created villain:", ebonite_mew)
print("Created villain:", dark_shorty)
print("Created villain:", ultra_bot)
ultra_bot.boss = thinnus
session.add(ultra_bot)
session.commit()
session.refresh(ultra_bot)
print("Updated villain:", ultra_bot)
clone_bot_1 = Villain(name="Clone Bot 1", power_level=2**6)
clone_bot_2 = Villain(name="Clone Bot 2", power_level=2**6)
clone_bot_3 = Villain(name="Clone Bot 3", power_level=2**6)
ultra_bot.minions.extend([clone_bot_1, clone_bot_2, clone_bot_3])
session.add(ultra_bot)
session.commit()
session.refresh(clone_bot_1)
session.refresh(clone_bot_2)
session.refresh(clone_bot_3)
print("Added minion:", clone_bot_1)
print("Added minion:", clone_bot_2)
print("Added minion:", clone_bot_3)
def main() -> None:
create_db_and_tables()
create_villains()
if __name__ == "__main__":
main()
Just as with regular relationships, we can simply pass our boss villain as an argument to the constructor using boss=thinnus.
If we later learn that a villain actually had a secret boss after we've already created him, we can just as easily assign that boss retroactively:
# Code above omitted 👆
def create_villains() -> None:
with Session(engine) as session:
# Code here omitted 👈
ultra_bot.boss = thinnus
session.add(ultra_bot)
session.commit()
session.refresh(ultra_bot)
print("Updated villain:", ultra_bot)
# Code below omitted 👇
👀 Full file preview
from typing import List, Optional
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine
class Villain(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
power_level: int
boss_id: Optional[int] = Field(
foreign_key="villain.id", default=None, nullable=True
)
boss: Optional["Villain"] = Relationship(
back_populates="minions",
sa_relationship_kwargs={"remote_side": "Villain.id"},
)
minions: List["Villain"] = Relationship(back_populates="boss")
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=False)
def create_db_and_tables() -> None:
SQLModel.metadata.create_all(engine)
def create_villains() -> None:
with Session(engine) as session:
thinnus = Villain(name="Thinnus", power_level=9001)
ebonite_mew = Villain(name="Ebonite Mew", power_level=400, boss=thinnus)
dark_shorty = Villain(name="Dark Shorty", power_level=200, boss=thinnus)
ultra_bot = Villain(name="Ultra Bot", power_level=2**9)
session.add(ebonite_mew)
session.add(dark_shorty)
session.add(ultra_bot)
session.commit()
session.refresh(thinnus)
session.refresh(ebonite_mew)
session.refresh(dark_shorty)
session.refresh(ultra_bot)
print("Created villain:", thinnus)
print("Created villain:", ebonite_mew)
print("Created villain:", dark_shorty)
print("Created villain:", ultra_bot)
ultra_bot.boss = thinnus
session.add(ultra_bot)
session.commit()
session.refresh(ultra_bot)
print("Updated villain:", ultra_bot)
clone_bot_1 = Villain(name="Clone Bot 1", power_level=2**6)
clone_bot_2 = Villain(name="Clone Bot 2", power_level=2**6)
clone_bot_3 = Villain(name="Clone Bot 3", power_level=2**6)
ultra_bot.minions.extend([clone_bot_1, clone_bot_2, clone_bot_3])
session.add(ultra_bot)
session.commit()
session.refresh(clone_bot_1)
session.refresh(clone_bot_2)
session.refresh(clone_bot_3)
print("Added minion:", clone_bot_1)
print("Added minion:", clone_bot_2)
print("Added minion:", clone_bot_3)
def main() -> None:
create_db_and_tables()
create_villains()
if __name__ == "__main__":
main()
And if we want to add minions to a boss afterward, it's as easy as adding items to a Python list (because that's all it is 🤓):
# Code above omitted 👆
def create_villains() -> None:
with Session(engine) as session:
# Code here omitted 👈
clone_bot_1 = Villain(name="Clone Bot 1", power_level=2**6)
clone_bot_2 = Villain(name="Clone Bot 2", power_level=2**6)
clone_bot_3 = Villain(name="Clone Bot 3", power_level=2**6)
ultra_bot.minions.extend([clone_bot_1, clone_bot_2, clone_bot_3])
session.add(ultra_bot)
session.commit()
session.refresh(clone_bot_1)
session.refresh(clone_bot_2)
session.refresh(clone_bot_3)
print("Added minion:", clone_bot_1)
print("Added minion:", clone_bot_2)
print("Added minion:", clone_bot_3)
# Code below omitted 👇
👀 Full file preview
from typing import List, Optional
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine
class Villain(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
power_level: int
boss_id: Optional[int] = Field(
foreign_key="villain.id", default=None, nullable=True
)
boss: Optional["Villain"] = Relationship(
back_populates="minions",
sa_relationship_kwargs={"remote_side": "Villain.id"},
)
minions: List["Villain"] = Relationship(back_populates="boss")
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=False)
def create_db_and_tables() -> None:
SQLModel.metadata.create_all(engine)
def create_villains() -> None:
with Session(engine) as session:
thinnus = Villain(name="Thinnus", power_level=9001)
ebonite_mew = Villain(name="Ebonite Mew", power_level=400, boss=thinnus)
dark_shorty = Villain(name="Dark Shorty", power_level=200, boss=thinnus)
ultra_bot = Villain(name="Ultra Bot", power_level=2**9)
session.add(ebonite_mew)
session.add(dark_shorty)
session.add(ultra_bot)
session.commit()
session.refresh(thinnus)
session.refresh(ebonite_mew)
session.refresh(dark_shorty)
session.refresh(ultra_bot)
print("Created villain:", thinnus)
print("Created villain:", ebonite_mew)
print("Created villain:", dark_shorty)
print("Created villain:", ultra_bot)
ultra_bot.boss = thinnus
session.add(ultra_bot)
session.commit()
session.refresh(ultra_bot)
print("Updated villain:", ultra_bot)
clone_bot_1 = Villain(name="Clone Bot 1", power_level=2**6)
clone_bot_2 = Villain(name="Clone Bot 2", power_level=2**6)
clone_bot_3 = Villain(name="Clone Bot 3", power_level=2**6)
ultra_bot.minions.extend([clone_bot_1, clone_bot_2, clone_bot_3])
session.add(ultra_bot)
session.commit()
session.refresh(clone_bot_1)
session.refresh(clone_bot_2)
session.refresh(clone_bot_3)
print("Added minion:", clone_bot_1)
print("Added minion:", clone_bot_2)
print("Added minion:", clone_bot_3)
def main() -> None:
create_db_and_tables()
create_villains()
if __name__ == "__main__":
main()
Since our relationships work both ways, we don't even need to add all our clone_bot_s to the session individually. Instead, we can simply add ultra_bot again and commit the changes. We do need to refresh them individually, though, if we want to access their updated attributes.
Traversing the relationship graph¶
By setting up our relationships this way, we can easily go back and forth along the graph representing all the relationships we've created so far.
For example, we can verify that our clone_bot_1 has a boss, who has his own boss, and that one of that top boss's minions is ebonite_mew:
top_boss_minions = clone_bot_3.boss.boss.minions
assert any(minion is ebonite_mew for minion in top_boss_minions) # passes
Info
Notice that we can, in fact, check for identity using is instead of == here, since we are dealing with the exact same objects, not just objects containing the same data.