I am trying to create two tables using SQLModel where a parent can have multiple children. The parents and their children will be found and registered to the database individually, so I need to somehow implement the ability to point already existing entries in the parent table to their children as they are found and registered, and vice-versa.
I have tried coming up with a proof-of-concept to register the parents and their children separately, and map the relationships as they are registered. I am running this using FastAPI, so the installation of the Python environment I used is as follows:
# Setting up the test Python environment
$ mamba/conda create my-sql-db python==3.9.*
$ mamba/conda activate my-sql-db
$ python -m pip install sqlmodel fastapi "uvicorn[standard]"
$ uvicorn db:app --reload # Run the FastAPI applet using Uvicorn
This is the standalone script to generate the two API endpoints I am trying to implement, with the script being named db.py:
from contextlib import asynccontextmanager
from sqlmodel import SQLModel, Session, Field, Relationship, create_engine, select
from fastapi import FastAPI
from typing import Optional, List, Union
"""
Set up example database to show the relationships I'm trying to establish
"""
# Define tables and relationships
class Parent(SQLModel, table=True): # type: ignore
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field() # In case more parameters are needed
children: List["Child"] = Relationship(
back_populates="parent",
sa_relationship_kwargs={"cascade": "delete"},
)
class Child(SQLModel, table=True): # type: ignore
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field() # In case more parameters are needed
parent: Optional["Parent"] = Relationship(
back_populates="children",
)
parent_id: Optional[int] = Field(
foreign_key="parent.id",
default=None,
)
# Set up database URL
db_name = "database.db"
db_url = f"sqlite:///{db_name}"
connect_args = {"check_same_thread": False}
engine = create_engine(db_url, echo=True, connect_args=connect_args)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
"""
Set up FastAPI app to interact with the database through
"""
app = FastAPI()
# Initialise database on FastAPI startup
@app.on_event("startup")
def on_startup():
create_db_and_tables()
# Helper functions
def check_db_entry_exists(
table: Union[Parent, Child],
name: str,
) -> bool:
"""
Checks if the name has already been registered in the database. Used to avoid registering duplicates
"""
with Session(engine) as session:
try:
session.exec(
select(table)
.where(table.name == name)
).one()
return True
except Exception:
return False
@app.post("/parent/")
def create_parent(
name: str,
children: List[str] = [],
) -> Parent:
"""
With this API endpoint, I am hoping to be able to add new parents to the database,
but also revisit and update an existing parent when a new child entry gets added.
"""
with Session(engine) as session:
# Check if the parent already exists
if check_db_entry_exists(Parent, name) is True:
parent = session.exec(
select(Parent)
.where(Parent.name == name)
).one()
# Eventually, I want to be able to check if info associated with the parent
# has changed, and to update the entry if it has
# Register the parent if they're new
else:
parent = Parent()
# Add name
parent.name = name
# Add children
if len(children) < 1:
pass
else:
for child in children:
parent.children.append(Child(name=child))
# Add to database and show
session.add(parent)
session.commit()
session.refresh(parent)
return parent
@app.post("/child/")
def create_child(
name: str,
parent: Optional[str] = None,
) -> Child:
"""
With this table, I'm hoping to add the child as a standalone entry first, then come
back and point it to a newly added parent once one gets added to the database
"""
with Session(engine) as session:
# Check if the child being registered already exists
if check_db_entry_exists(Child, name) is True:
child = session.exec(
select(Child)
.where(Child.name == name)
).one()
# Eventually, I want to be able to check if the information associated with
# the child has changed, and update it if it has
# Register child to the database if it doesn't already exist
else:
child = Child()
# Add name
child.name = name
# Add parent
child.parent = Parent(name=parent)
# Add to database and show
session.add(child)
session.commit()
session.refresh(child)
return child
At present, when I add a new entry to the child table via the FastAPI docs page and try to point it to the parent, a duplicate of the parent gets created instead of using an already existing entry. I'm new to working with SQL databases and SQLModel, so could you tell me how I can modify/add to what I've already got in order to implement what I've described?
In your condition where the child doesn't exist you want to query the database for the parent on whatever condition is appropriate. If there is an existing record, use that, otherwise create a new Parent object and assign that. Something along the lines of:
...
# Register child to the database if it doesn't already exist
else:
parent_to_link = db.query(Parent).filter(Parent.name == parent).first()
if not parent_to_link:
parent_to_link= Parent(name=parent)
# ...
child.parent = parent_to_link