Description
\nI am trying to work out what the best method is for creating SQLModel objects with nested objects beneath them.
\nThe example code in the tutorial shows reading data (such as with HeroPublicWithTeam) but not creating data.
\nMy code above seems to work ok - it is just the Models with Relationships in FastAPI code with the HeroCreate class modified to optionally allow a Team to also be created within the same API call - and team_id moved from HeroBase to Hero as it doesn't make sense for HeroCreate to have a team and a team_id in it.
Is this the best way to achieve this?
\nOr am I missing a simpler way to do this that doesn't involve an AfterValidator which seems like an odd way to achieve this?
\nJust having team: TeamCreate | None = None leads to 'TeamCreate' object has no attribute '_sa_instance_state'
\nJust having team: Team | None = None means that id is part of the API for creating a team within the hero creation leading to the inevitable sqlalchemy.exc.IntegrityError: (psycopg.errors.UniqueViolation) duplicate key value violates unique constraint \"team_pkey\"
Operating System
\nLinux
\nOperating System Details
\npython-slim docker container
\nSQLModel Version
\n0.0.22
\nPython Version
\n3.12.6
\nAdditional Context
\nNo response
","upvoteCount":3,"answerCount":2,"acceptedAnswer":{"@type":"Answer","text":"I ran into the same issue and was inspired by this thread. Here’s my refinement:
\nI introduced Helper models that sit between the Create schemas and the table=True ORM models. These helpers use @field_validator(..., mode=\"after\") to recursively convert nested data into ORM-compatible objects — without mutating the original input schemas or embedding transformation logic into the ORM models.
This approach keeps input validation clean while making the transformation layer explicit, type-safe, and scalable.
\n✅ Minimal working example
\nfrom sqlmodel import SQLModel, Field, Relationship\nfrom pydantic import field_validator\n\n# --- Input Schemas ---\nclass AddressCreate(SQLModel):\n city: str\n\nclass UserCreate(SQLModel):\n name: str\n address: AddressCreate\n\n# --- ORM Models ---\nclass AddressModel(SQLModel, table=True):\n id: int | None = Field(default=None, primary_key=True)\n city: str\n\nclass UserModel(SQLModel, table=True):\n id: int | None = Field(default=None, primary_key=True)\n name: str\n address_id: int | None = Field(default=None, foreign_key=\"addressmodel.id\")\n address: AddressModel = Relationship()\n\n# --- Helper Models ---\nclass AddressCreateHelper(AddressCreate):\n @field_validator(\"city\", mode=\"after\")\n @classmethod\n def to_model(cls, v, values):\n return AddressModel(city=v)\n\nclass UserCreateHelper(UserCreate):\n @field_validator(\"address\", mode=\"after\")\n @classmethod\n def to_model(cls, v):\n return AddressModel.model_validate(AddressCreateHelper.model_validate(v))\n\n# --- Usage ---\npayload = {\"name\": \"Alice\", \"address\": {\"city\": \"Berlin\"}}\n\n# In the FastAPI route or controller layer\nuser_input = UserCreate.model_validate(payload)\n\n# In the transformation layer (e.g., service or use case layer)\nuser_helper = UserCreateHelper.model_validate(user_input)\n\n# In the persistence layer\nuser_model = UserModel.model_validate(user_helper)This keeps UserCreate, UserModel, and UserCreateHelper cleanly separated for API input, database persistence, and transformation logic — making the system easier to maintain and reason about.
Best way to add nested model with relationship in Create #1116
-
First Check
Commit to Help
Example Codefrom fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select
from typing import Annotated
from pydantic import AfterValidator
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 HeroBase(SQLModel):
name: str = Field(index=True)
secret_name: str
age: int | None = Field(default=None, index=True)
class Hero(HeroBase, table=True):
id: int | None = Field(default=None, primary_key=True)
team_id: int | None = Field(default=None, foreign_key="team.id")
team: Team | None = Relationship(back_populates="heroes")
class HeroPublic(HeroBase):
id: int
class HeroCreate(HeroBase):
team: (
Annotated[
TeamCreate,
AfterValidator(Team.model_validate),
]
| None
) = None
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: TeamPublic | None = None
class TeamPublicWithHeroes(TeamPublic):
heroes: list[HeroPublic] = []
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}DescriptionI am trying to work out what the best method is for creating SQLModel objects with nested objects beneath them. Is this the best way to achieve this? Operating SystemLinux Operating System Detailspython-slim docker container SQLModel Version0.0.22 Python Version3.12.6 Additional ContextNo response |
Beta Was this translation helpful? Give feedback.
Replies: 2 comments 2 replies
-
|
Hi, @jonyscathe I tried your solution and it also worked for me, but there is still a thing to consider. At the first line of a path operation, when fastapi validation is finished and before Hero itself has a I am not sure where, but I have an intuition, that this inconsistency could cause a bug down a stream. I was thinking to write a Another option is to not perform any automatic conversions and write all by hands. This approach is recommended in DRF, for example:
At last, what is a need to perform an object creation in path operation code? This data is more likely to be passed to some service containing a business logic, which handles connected objects creation. So for myself I would either call pop teams info and call PS: another catch - if you pass a foreign key of non-existing key while creating a book - it will return 500 instead of 400 with a fancy error description, because 666 is still a valid int, although team with id 666 does not exist. |
Beta Was this translation helpful? Give feedback.
-
|
I ran into the same issue and was inspired by this thread. Here’s my refinement: I introduced Helper models that sit between the This approach keeps input validation clean while making the transformation layer explicit, type-safe, and scalable. ✅ Minimal working examplefrom sqlmodel import SQLModel, Field, Relationship
from pydantic import field_validator
# --- Input Schemas ---
class AddressCreate(SQLModel):
city: str
class UserCreate(SQLModel):
name: str
address: AddressCreate
# --- ORM Models ---
class AddressModel(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
city: str
class UserModel(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
address_id: int | None = Field(default=None, foreign_key="addressmodel.id")
address: AddressModel = Relationship()
# --- Helper Models ---
class AddressCreateHelper(AddressCreate):
@field_validator("city", mode="after")
@classmethod
def to_model(cls, v, values):
return AddressModel(city=v)
class UserCreateHelper(UserCreate):
@field_validator("address", mode="after")
@classmethod
def to_model(cls, v):
return AddressModel.model_validate(AddressCreateHelper.model_validate(v))
# --- Usage ---
payload = {"name": "Alice", "address": {"city": "Berlin"}}
# In the FastAPI route or controller layer
user_input = UserCreate.model_validate(payload)
# In the transformation layer (e.g., service or use case layer)
user_helper = UserCreateHelper.model_validate(user_input)
# In the persistence layer
user_model = UserModel.model_validate(user_helper)This keeps |
Beta Was this translation helpful? Give feedback.
-
|
Nice solution. I ended up doing something kinda similar (though not as neat) and never updated this thread so I am glad someone did. |
Beta Was this translation helpful? Give feedback.
-
|
Hi @gzu300 , would you like to share the rest of the script with the path operations? |
Beta Was this translation helpful? Give feedback.
I ran into the same issue and was inspired by this thread. Here’s my refinement:
I introduced Helper models that sit between the
Createschemas and thetable=TrueORM models. These helpers use@field_validator(..., mode="after")to recursively convert nested data into ORM-compatible objects — without mutating the original input schemas or embedding transformation logic into the ORM models.This approach keeps input validation clean while making the transformation layer explicit, type-safe, and scalable.
✅ Minimal working example