First, why is it that we are not getting the related data for each hero and for each team?
It's because we declared the HeroRead with only the same base fields of the HeroBase plus the id. But it doesn't include a field team for the relationship attribute.
And the same way, we declared the TeamRead with only the same base fields of the TeamBase plus the id. But it doesn't include a field heroes for the relationship attribute.
# Code above omitted πclassTeamBase(SQLModel):name:str=Field(index=True)headquarters:str# Code here omitted πclassTeamRead(TeamBase):id:int# Code here omitted πclassHeroBase(SQLModel):name:str=Field(index=True)secret_name:strage:Optional[int]=Field(default=None,index=True)team_id:Optional[int]=Field(default=None,foreign_key="team.id")# Code here omitted πclassHeroRead(HeroBase):id:int# Code below omitted π
π Full file preview
fromtypingimportList,OptionalfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Relationship,Session,SQLModel,create_engine,selectclassTeamBase(SQLModel):name:str=Field(index=True)headquarters:strclassTeam(TeamBase,table=True):id:Optional[int]=Field(default=None,primary_key=True)heroes:List["Hero"]=Relationship(back_populates="team")classTeamCreate(TeamBase):passclassTeamRead(TeamBase):id:intclassTeamUpdate(SQLModel):name:Optional[str]=Noneheadquarters:Optional[str]=NoneclassHeroBase(SQLModel):name:str=Field(index=True)secret_name:strage:Optional[int]=Field(default=None,index=True)team_id:Optional[int]=Field(default=None,foreign_key="team.id")classHero(HeroBase,table=True):id:Optional[int]=Field(default=None,primary_key=True)team:Optional[Team]=Relationship(back_populates="heroes")classHeroRead(HeroBase):id:intclassHeroCreate(HeroBase):passclassHeroUpdate(SQLModel):name:Optional[str]=Nonesecret_name:Optional[str]=Noneage:Optional[int]=Noneteam_id:Optional[int]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,echo=True,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroRead)defcreate_hero(*,session:Session=Depends(get_session),hero:HeroCreate):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroRead])defread_heroes(*,session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroRead)defread_hero(*,session:Session=Depends(get_session),hero_id:int):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroRead)defupdate_hero(*,session:Session=Depends(get_session),hero_id:int,hero:HeroUpdate):db_hero=session.get(Hero,hero_id)ifnotdb_hero:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)forkey,valueinhero_data.items():setattr(db_hero,key,value)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.delete("/heroes/{hero_id}")defdelete_hero(*,session:Session=Depends(get_session),hero_id:int):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}@app.post("/teams/",response_model=TeamRead)defcreate_team(*,session:Session=Depends(get_session),team:TeamCreate):db_team=Team.model_validate(team)session.add(db_team)session.commit()session.refresh(db_team)returndb_team@app.get("/teams/",response_model=List[TeamRead])defread_teams(*,session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):teams=session.exec(select(Team).offset(offset).limit(limit)).all()returnteams@app.get("/teams/{team_id}",response_model=TeamRead)defread_team(*,team_id:int,session:Session=Depends(get_session)):team=session.get(Team,team_id)ifnotteam:raiseHTTPException(status_code=404,detail="Team not found")returnteam@app.patch("/teams/{team_id}",response_model=TeamRead)defupdate_team(*,session:Session=Depends(get_session),team_id:int,team:TeamUpdate,):db_team=session.get(Team,team_id)ifnotdb_team:raiseHTTPException(status_code=404,detail="Team not found")team_data=team.model_dump(exclude_unset=True)forkey,valueinteam_data.items():setattr(db_team,key,value)session.add(db_team)session.commit()session.refresh(db_team)returndb_team@app.delete("/teams/{team_id}")defdelete_team(*,session:Session=Depends(get_session),team_id:int):team=session.get(Team,team_id)ifnotteam:raiseHTTPException(status_code=404,detail="Team not found")session.delete(team)session.commit()return{"ok":True}
In this case, we used response_model=TeamRead and response_model=HeroRead, so FastAPI will use them to filter the response data, even if we return a table model that includes relationship attributes:
# Code above omitted π@app.get("/heroes/{hero_id}",response_model=HeroRead)defread_hero(*,session:Session=Depends(get_session),hero_id:int):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero# Code here omitted π@app.get("/teams/{team_id}",response_model=TeamRead)defread_team(*,team_id:int,session:Session=Depends(get_session)):team=session.get(Team,team_id)ifnotteam:raiseHTTPException(status_code=404,detail="Team not found")returnteam# Code below omitted π
π Full file preview
fromtypingimportList,OptionalfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Relationship,Session,SQLModel,create_engine,selectclassTeamBase(SQLModel):name:str=Field(index=True)headquarters:strclassTeam(TeamBase,table=True):id:Optional[int]=Field(default=None,primary_key=True)heroes:List["Hero"]=Relationship(back_populates="team")classTeamCreate(TeamBase):passclassTeamRead(TeamBase):id:intclassTeamUpdate(SQLModel):name:Optional[str]=Noneheadquarters:Optional[str]=NoneclassHeroBase(SQLModel):name:str=Field(index=True)secret_name:strage:Optional[int]=Field(default=None,index=True)team_id:Optional[int]=Field(default=None,foreign_key="team.id")classHero(HeroBase,table=True):id:Optional[int]=Field(default=None,primary_key=True)team:Optional[Team]=Relationship(back_populates="heroes")classHeroRead(HeroBase):id:intclassHeroCreate(HeroBase):passclassHeroUpdate(SQLModel):name:Optional[str]=Nonesecret_name:Optional[str]=Noneage:Optional[int]=Noneteam_id:Optional[int]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,echo=True,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroRead)defcreate_hero(*,session:Session=Depends(get_session),hero:HeroCreate):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroRead])defread_heroes(*,session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroRead)defread_hero(*,session:Session=Depends(get_session),hero_id:int):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroRead)defupdate_hero(*,session:Session=Depends(get_session),hero_id:int,hero:HeroUpdate):db_hero=session.get(Hero,hero_id)ifnotdb_hero:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)forkey,valueinhero_data.items():setattr(db_hero,key,value)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.delete("/heroes/{hero_id}")defdelete_hero(*,session:Session=Depends(get_session),hero_id:int):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}@app.post("/teams/",response_model=TeamRead)defcreate_team(*,session:Session=Depends(get_session),team:TeamCreate):db_team=Team.model_validate(team)session.add(db_team)session.commit()session.refresh(db_team)returndb_team@app.get("/teams/",response_model=List[TeamRead])defread_teams(*,session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):teams=session.exec(select(Team).offset(offset).limit(limit)).all()returnteams@app.get("/teams/{team_id}",response_model=TeamRead)defread_team(*,team_id:int,session:Session=Depends(get_session)):team=session.get(Team,team_id)ifnotteam:raiseHTTPException(status_code=404,detail="Team not found")returnteam@app.patch("/teams/{team_id}",response_model=TeamRead)defupdate_team(*,session:Session=Depends(get_session),team_id:int,team:TeamUpdate,):db_team=session.get(Team,team_id)ifnotdb_team:raiseHTTPException(status_code=404,detail="Team not found")team_data=team.model_dump(exclude_unset=True)forkey,valueinteam_data.items():setattr(db_team,key,value)session.add(db_team)session.commit()session.refresh(db_team)returndb_team@app.delete("/teams/{team_id}")defdelete_team(*,session:Session=Depends(get_session),team_id:int):team=session.get(Team,team_id)ifnotteam:raiseHTTPException(status_code=404,detail="Team not found")session.delete(team)session.commit()return{"ok":True}
We cannot simply include all the data, including all the internal relationships, because each hero has an attribute team with their team, and then that team also has an attribute heroes with all the heroes in the team, including this one.
If we tried to include everything, we could make the server application crash trying to extract infinite data, going through the same hero and team over and over again internally, something like this:
As you can see, in this example, we would get the hero Rusty-Man, and from this hero we would get the team Preventers, and then from this team we would get its heroes, of course, including Rusty-Man... π±
So we start again, and in the end, the server would just crash trying to get all the data with a "Maximum recursion error", we would not even get a response like the one above.
So, we need to carefully choose in which cases we want to include data and in which not.
fromtypingimportList,OptionalfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Relationship,Session,SQLModel,create_engine,selectclassTeamBase(SQLModel):name:str=Field(index=True)headquarters:strclassTeam(TeamBase,table=True):id:Optional[int]=Field(default=None,primary_key=True)heroes:List["Hero"]=Relationship(back_populates="team")classTeamCreate(TeamBase):passclassTeamRead(TeamBase):id:intclassTeamUpdate(SQLModel):id:Optional[int]=Nonename:Optional[str]=Noneheadquarters:Optional[str]=NoneclassHeroBase(SQLModel):name:str=Field(index=True)secret_name:strage:Optional[int]=Field(default=None,index=True)team_id:Optional[int]=Field(default=None,foreign_key="team.id")classHero(HeroBase,table=True):id:Optional[int]=Field(default=None,primary_key=True)team:Optional[Team]=Relationship(back_populates="heroes")classHeroRead(HeroBase):id:intclassHeroCreate(HeroBase):passclassHeroUpdate(SQLModel):name:Optional[str]=Nonesecret_name:Optional[str]=Noneage:Optional[int]=Noneteam_id:Optional[int]=NoneclassHeroReadWithTeam(HeroRead):team:Optional[TeamRead]=NoneclassTeamReadWithHeroes(TeamRead):heroes:List[HeroRead]=[]sqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,echo=True,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroRead)defcreate_hero(*,session:Session=Depends(get_session),hero:HeroCreate):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroRead])defread_heroes(*,session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroReadWithTeam)defread_hero(*,session:Session=Depends(get_session),hero_id:int):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroRead)defupdate_hero(*,session:Session=Depends(get_session),hero_id:int,hero:HeroUpdate):db_hero=session.get(Hero,hero_id)ifnotdb_hero:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)forkey,valueinhero_data.items():setattr(db_hero,key,value)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.delete("/heroes/{hero_id}")defdelete_hero(*,session:Session=Depends(get_session),hero_id:int):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}@app.post("/teams/",response_model=TeamRead)defcreate_team(*,session:Session=Depends(get_session),team:TeamCreate):db_team=Team.model_validate(team)session.add(db_team)session.commit()session.refresh(db_team)returndb_team@app.get("/teams/",response_model=List[TeamRead])defread_teams(*,session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):teams=session.exec(select(Team).offset(offset).limit(limit)).all()returnteams@app.get("/teams/{team_id}",response_model=TeamReadWithHeroes)defread_team(*,team_id:int,session:Session=Depends(get_session)):team=session.get(Team,team_id)ifnotteam:raiseHTTPException(status_code=404,detail="Team not found")returnteam@app.patch("/teams/{team_id}",response_model=TeamRead)defupdate_team(*,session:Session=Depends(get_session),team_id:int,team:TeamUpdate,):db_team=session.get(Team,team_id)ifnotdb_team:raiseHTTPException(status_code=404,detail="Team not found")team_data=team.model_dump(exclude_unset=True)forkey,valueinteam_data.items():setattr(db_team,key,value)session.add(db_team)session.commit()session.refresh(db_team)returndb_team@app.delete("/teams/{team_id}")defdelete_team(*,session:Session=Depends(get_session),team_id:int):team=session.get(Team,team_id)ifnotteam:raiseHTTPException(status_code=404,detail="Team not found")session.delete(team)session.commit()return{"ok":True}
These two models are very simple in code, but there's a lot happening here. Let's check it out.
The HeroReadWithTeaminherits from HeroRead, which means that it will have the normal fields for reading, including the required id that was declared in HeroRead.
And then it adds the new fieldteam, which could be None, and is declared with the type TeamRead with the base fields for reading a team.
Then we do the same for the TeamReadWithHeroes, it inherits from TeamRead, and declares the new fieldheroes, which is a list of HeroRead.
Now, notice that these new fields team and heroes are not declared with Relationship(), because these are not table models, they cannot have relationship attributes with the magic access to get that data from the database.
Instead, here these are only data models that will tell FastAPI which attributes to get data from and which data to get from them.
Also, notice that the field team is not declared with this new TeamReadWithHeroes, because that would again create that infinite recursion of data. Instead, we declare it with the normal TeamRead model.
And the same for TeamReadWithHeroes, the model used for the new field heroes uses HeroRead to get only each hero's data.
This also means that, even though we have these two new models, we still need the previous ones, HeroRead and TeamRead, because we need to reference them here (and we are also using them in the rest of the path operations).
Now we can update the path operations to use the new models.
This will tell FastAPI to take the object that we return from the path operation function (a table model) and access the additional attributes from them to extract their data.
In the case of the hero, this tells FastAPI to extract the team too. And in the case of the team, to extract the list of heroes too.
# Code above omitted π@app.get("/heroes/{hero_id}",response_model=HeroReadWithTeam)defread_hero(*,session:Session=Depends(get_session),hero_id:int):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero# Code here omitted π@app.get("/teams/{team_id}",response_model=TeamReadWithHeroes)defread_team(*,team_id:int,session:Session=Depends(get_session)):team=session.get(Team,team_id)ifnotteam:raiseHTTPException(status_code=404,detail="Team not found")returnteam# Code below omitted π
π Full file preview
fromtypingimportList,OptionalfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Relationship,Session,SQLModel,create_engine,selectclassTeamBase(SQLModel):name:str=Field(index=True)headquarters:strclassTeam(TeamBase,table=True):id:Optional[int]=Field(default=None,primary_key=True)heroes:List["Hero"]=Relationship(back_populates="team")classTeamCreate(TeamBase):passclassTeamRead(TeamBase):id:intclassTeamUpdate(SQLModel):id:Optional[int]=Nonename:Optional[str]=Noneheadquarters:Optional[str]=NoneclassHeroBase(SQLModel):name:str=Field(index=True)secret_name:strage:Optional[int]=Field(default=None,index=True)team_id:Optional[int]=Field(default=None,foreign_key="team.id")classHero(HeroBase,table=True):id:Optional[int]=Field(default=None,primary_key=True)team:Optional[Team]=Relationship(back_populates="heroes")classHeroRead(HeroBase):id:intclassHeroCreate(HeroBase):passclassHeroUpdate(SQLModel):name:Optional[str]=Nonesecret_name:Optional[str]=Noneage:Optional[int]=Noneteam_id:Optional[int]=NoneclassHeroReadWithTeam(HeroRead):team:Optional[TeamRead]=NoneclassTeamReadWithHeroes(TeamRead):heroes:List[HeroRead]=[]sqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,echo=True,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroRead)defcreate_hero(*,session:Session=Depends(get_session),hero:HeroCreate):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroRead])defread_heroes(*,session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroReadWithTeam)defread_hero(*,session:Session=Depends(get_session),hero_id:int):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroRead)defupdate_hero(*,session:Session=Depends(get_session),hero_id:int,hero:HeroUpdate):db_hero=session.get(Hero,hero_id)ifnotdb_hero:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)forkey,valueinhero_data.items():setattr(db_hero,key,value)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.delete("/heroes/{hero_id}")defdelete_hero(*,session:Session=Depends(get_session),hero_id:int):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}@app.post("/teams/",response_model=TeamRead)defcreate_team(*,session:Session=Depends(get_session),team:TeamCreate):db_team=Team.model_validate(team)session.add(db_team)session.commit()session.refresh(db_team)returndb_team@app.get("/teams/",response_model=List[TeamRead])defread_teams(*,session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):teams=session.exec(select(Team).offset(offset).limit(limit)).all()returnteams@app.get("/teams/{team_id}",response_model=TeamReadWithHeroes)defread_team(*,team_id:int,session:Session=Depends(get_session)):team=session.get(Team,team_id)ifnotteam:raiseHTTPException(status_code=404,detail="Team not found")returnteam@app.patch("/teams/{team_id}",response_model=TeamRead)defupdate_team(*,session:Session=Depends(get_session),team_id:int,team:TeamUpdate,):db_team=session.get(Team,team_id)ifnotdb_team:raiseHTTPException(status_code=404,detail="Team not found")team_data=team.model_dump(exclude_unset=True)forkey,valueinteam_data.items():setattr(db_team,key,value)session.add(db_team)session.commit()session.refresh(db_team)returndb_team@app.delete("/teams/{team_id}")defdelete_team(*,session:Session=Depends(get_session),team_id:int):team=session.get(Team,team_id)ifnotteam:raiseHTTPException(status_code=404,detail="Team not found")session.delete(team)session.commit()return{"ok":True}
Using the same techniques to declare additional data models, we can tell FastAPI what data to return in the responses, even when we return table models.
Here we almost didn't have to change the FastAPI app code, but of course, there will be cases where you need to get the data and process it in different ways in the path operation function before returning it.
But even in those cases, you will be able to define the data models to use in response_model to tell FastAPI how to validate and filter the data.
By this point, you already have a very robust API to handle data in a SQL database combining SQLModel with FastAPI, and implementing best practices, like data validation, conversion, filtering, and documentation. β¨
In the next chapter, I'll tell you how to implement automated testing for your application using FastAPI and SQLModel. β