FastAPI doesn't start when I split my models into multiple files. (Models with Relationships) #757
-
First Check
Commit to Help
Example Code############ hero.py file ############
from typing import Optional, TYPE_CHECKING
from sqlmodel import SQLModel, Field, Relationship
if TYPE_CHECKING:
from src.app.models.team import TeamRead, Team
class HeroBase(SQLModel):
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None, index=True)
team_id: Optional[int] = Field(default=None, foreign_key="team.id")
class Hero(HeroBase, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
team: Optional["Team"] = Relationship(back_populates="heroes")
class HeroRead(HeroBase):
id: int
class HeroCreate(HeroBase):
pass
class HeroUpdate(SQLModel):
name: Optional[str] = None
secret_name: Optional[str] = None
age: Optional[int] = None
team_id: Optional[int] = None
class HeroReadWithTeam(HeroRead):
team: Optional["TeamRead"] = None
############ team.py file############
from typing import Optional, List, TYPE_CHECKING
from sqlmodel import SQLModel, Field, Relationship
if TYPE_CHECKING:
from src.app.models.hero import HeroRead
class TeamBase(SQLModel):
name: str = Field(index=True)
headquarters: str
class Team(TeamBase, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
heroes: List["Hero"] = Relationship(back_populates="team")
class TeamCreate(TeamBase):
pass
class TeamRead(TeamBase):
id: int
class TeamUpdate(SQLModel):
id: Optional[int] = None
name: Optional[str] = None
headquarters: Optional[str] = None
class TeamReadWithHeroes(TeamRead):
heroes: List["HeroRead"] = []
######__init__.py file ######
from .hero import Hero
from .team import Team
##### api/teams.py #####
from typing import List
from fastapi import Depends, APIRouter, Query
from sqlmodel import Session, select
from src.app.db.session import get_session
from src.app.models.team import TeamRead, Team, TeamCreate, TeamReadWithHeroes
router = APIRouter()
@router.post("/teams/", response_model=TeamRead)
def create_team(*, session: Session = Depends(get_session), team: TeamCreate):
db_team = Team.model_validate(team)
session.add(db_team)
session.commit()
session.refresh(db_team)
return db_team
@router.get("/teams/", response_model=List[TeamReadWithHeroes])
def read_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()
return teams DescriptionAs in the example provided on the website, I want to use Models with Relationships in FastAPI. It works fine when I work within a single file. However, I'd like to split the models into different files. When I do so (as in the code example I provided), I encounter the following error upon execution:
The problem is in this code :
Has anyone else encountered this issue and can help me? Operating SystemmacOS Operating System DetailsSonoma 14.1.1 SQLModel Version0.0.14 Python Version3.12.0 Additional ContextFastAPI version : 0.105.0 |
Beta Was this translation helpful? Give feedback.
Replies: 3 comments 5 replies
-
I’ve encountered the same issue myself. |
Beta Was this translation helpful? Give feedback.
-
I've also been facing this issue. I found a workaround here, but it ain't pretty. You can move the circular imports to the bottom of the file and call ############ models/hero.py ############
from typing import Optional, TYPE_CHECKING
"""
More code here...
"""
class HeroReadWithTeam(HeroRead):
team: Optional["TeamRead"] = None
# At the bottom of the file
from .team import TeamRead
HeroReadWithTeam.model_rebuild()
############ models/team.py ############
from typing import Optional, List, TYPE_CHECKING
"""
More code here...
"""
class TeamReadWithHeroes(TeamRead):
heroes: List["HeroRead"] = []
# At the bottom of the file
from .hero import TeamReadWithHeroes
TeamReadWithHeroes.model_rebuild() Explanation: the first time Python tries to import from a file, it builds a In our case, Python is already aware of the models it needs from |
Beta Was this translation helpful? Give feedback.
-
Using:
And the examples provided in SQLModel Tutorial: Is this THE solution to go with or what would you recommend @tiangolo ? See this code, combining above mentioned 3 sources. This is a 100% working full example:
hero_model.py: from typing import TYPE_CHECKING, Optional
from sqlmodel import Field, Relationship, SQLModel
if TYPE_CHECKING:
from team_model import Team, TeamPublic
class HeroBase(SQLModel):
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")
class Hero(HeroBase, table=True):
id: int | None = Field(default=None, primary_key=True)
team: Optional["Team"] = Relationship(back_populates="heroes")
class HeroPublic(HeroBase):
id: int
class HeroCreate(HeroBase):
pass
class HeroUpdate(SQLModel):
name: str | None = None
secret_name: str | None = None
age: int | None = None
team_id: int | None = None
class HeroPublicWithTeam(HeroPublic):
team: Optional["TeamPublic"] = None
# fix 'PydanticUndefinedAnnotation: name 'TeamPublic' is not defined' error
# see: https://github.com/tiangolo/sqlmodel/discussions/757
from team_model import TeamPublic
HeroPublicWithTeam.model_rebuild() team_model.py: from typing import TYPE_CHECKING
from sqlmodel import Field, Relationship, SQLModel
if TYPE_CHECKING:
from hero_model import Hero, HeroPublic
class TeamBase(SQLModel):
name: str = Field(index=True)
headquarters: str
class Team(TeamBase, table=True):
id: int | None = Field(default=None, primary_key=True)
heroes: list["Hero"] = Relationship(back_populates="team")
class TeamCreate(TeamBase):
pass
class TeamPublic(TeamBase):
id: int
class TeamUpdate(SQLModel):
id: int | None = None
name: str | None = None
headquarters: str | None = None
class TeamPublicWithHeroes(TeamPublic):
heroes: list["HeroPublic"] = []
# fix 'PydanticUndefinedAnnotation: name 'HeroPublic' is not defined' error
# see: https://github.com/tiangolo/sqlmodel/discussions/757
from hero_model import HeroPublic
TeamPublicWithHeroes.model_rebuild() main.py: #!/usr/bin/env python
from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Session, SQLModel, create_engine, select
import uvicorn
from hero_model import Hero, HeroCreate, HeroPublic, HeroPublicWithTeam, HeroUpdate
from team_model import Team, TeamCreate, TeamPublic, TeamPublicWithHeroes, TeamUpdate
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)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def get_session():
with Session(engine) as session:
yield session
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate):
db_hero = Hero.model_validate(hero)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.get("/heroes/", response_model=list[HeroPublic])
def read_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()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroPublicWithTeam)
def read_hero(*, session: Session = Depends(get_session), hero_id: int):
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(*, session: Session = Depends(get_session), hero_id: int, hero: HeroUpdate):
db_hero = session.get(Hero, hero_id)
if not db_hero:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
for key, value in hero_data.items():
setattr(db_hero, key, value)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.delete("/heroes/{hero_id}")
def delete_hero(*, session: Session = Depends(get_session), hero_id: int):
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
session.delete(hero)
session.commit()
return {"ok": True}
@app.post("/teams/", response_model=TeamPublic)
def create_team(*, session: Session = Depends(get_session), team: TeamCreate):
db_team = Team.model_validate(team)
session.add(db_team)
session.commit()
session.refresh(db_team)
return db_team
@app.get("/teams/", response_model=list[TeamPublic])
def read_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()
return teams
@app.get("/teams/{team_id}", response_model=TeamPublicWithHeroes)
def read_team(*, team_id: int, session: Session = Depends(get_session)):
team = session.get(Team, team_id)
if not team:
raise HTTPException(status_code=404, detail="Team not found")
return team
@app.patch("/teams/{team_id}", response_model=TeamPublic)
def update_team(
*,
session: Session = Depends(get_session),
team_id: int,
team: TeamUpdate,
):
db_team = session.get(Team, team_id)
if not db_team:
raise HTTPException(status_code=404, detail="Team not found")
team_data = team.model_dump(exclude_unset=True)
for key, value in team_data.items():
setattr(db_team, key, value)
session.add(db_team)
session.commit()
session.refresh(db_team)
return db_team
@app.delete("/teams/{team_id}")
def delete_team(*, session: Session = Depends(get_session), team_id: int):
team = session.get(Team, team_id)
if not team:
raise HTTPException(status_code=404, detail="Team not found")
session.delete(team)
session.commit()
return {"ok": True}
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=8888) |
Beta Was this translation helpful? Give feedback.
Hi guys, I just found another solution that worked on my side.
You can create a
models
package where you put all you model files likehero.py
andteam.py
. In this package, use the__init__.py
to import your public models and usemodel_rebuild()
in it. Using this solution you'll no longer have to import and rebuild at the end of each file.