Return Nested JSON in FastAPI with SQLModel: Replace join rows with relationships and Pydantic response models

Learn how to fix FastAPI and SQLModel nested response validation errors from join results. Build relationships and Pydantic schemas for clean, documented JSON.

Deeply nested response shapes in FastAPI with SQLModel often look straightforward on paper, but in practice the first join-based attempt tends to hit a wall at response validation. If you’ve wired a join and got back proper rows, yet the OpenAPI console shows either nulls or hard-to-read validation traces, the issue is not in the database. It’s in the mismatch between what your endpoint returns and what the response model expects.

Problem, reproduced

Consider a simple schema of attacks and their options. The goal is to return a response that nests options inside each attack. The initial implementation looks almost correct, but it returns raw join rows instead of a properly structured object.

from typing import Annotated
import logging
from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select
log = logging.getLogger("uvicorn.error")
class Exploit(SQLModel, table=True):
    __tablename__ = "attacks_attack"
    attack_id: int | None = Field(default=None, primary_key=True)
    name: str
    module: str
class Opt(SQLModel, table=True):
    __tablename__ = "attacks_option"
    option_id: int | None = Field(default=None, primary_key=True)
    name: str
    attack_id: int = Field(default=None, foreign_key="attacks_attack.attack_id")
class ExploitWithOpts(SQLModel):
    attack_id: int | None = Field(default=None, primary_key=True)
    name: str
    module: str
    options: tuple[Opt]
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)
def init_db():
    SQLModel.metadata.create_all(engine)
def session_provider():
    with Session(engine) as s:
        yield s
DBSess = Annotated[Session, Depends(session_provider)]
app = FastAPI()
@app.get("/attacks/")
def fetch_exploits(
    db: DBSess,
    offset: int = 0,
    limit: Annotated[int, Query(le=100)] = 1,
) -> list[ExploitWithOpts]:
    rows = db.exec(select(Exploit, Opt).join(Opt).offset(offset).limit(limit)).all()
    log.info(rows)
    return rows

Why it fails

The join produces a list of tuples like (Exploit, Opt). That is a perfectly fine SQL result, but it does not match the response model. The response model declares a list of objects with top-level fields attack_id, name, module and a nested collection of options. FastAPI hands that list of tuples to Pydantic, which tries to validate each tuple against ExploitWithOpts and fails with “Field required” for name and module, and later with “Input should be a valid dictionary” when the return type drifts further away from the declared model.

A second attempt to bolt a relationship attribute onto the ad-hoc response model doesn’t help either. Relationship belongs to table models designed for ORM navigation, not to arbitrary response containers. The result is still rows, not a nested object, and the validator is still unhappy.

Working approach

The robust pattern here is to model relationships on table models, define explicit response schemas for the nested JSON you want, and then assemble that JSON from ORM instances. The shapes that go over the wire should not be raw join rows. They should be dictionaries or Pydantic models that mirror your intended API.

The implementation below wires relationships on table models, declares response schemas, and builds the nested payloads, module options, and targets by iterating over relationships.

# models/vector.py
from typing import List, Optional, TYPE_CHECKING
from sqlmodel import SQLModel, Field, Relationship
from .links import VectorArtifactLink
if TYPE_CHECKING:
    from .artifact import Artifact
    from .opts import ModOptGroup
    from .aim import Aim
class Vector(SQLModel, table=True):
    __tablename__ = "attacks_attack"
    attack_id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    module: str
    platform: Optional[str]
    arch: Optional[str]
    privileged: Optional[str]
    license: Optional[str]
    rank: Optional[str]
    disclosed: Optional[str]
    provided_by: Optional[str]
    module_side_effects: Optional[str]
    module_stability: Optional[str]
    module_reliability: Optional[str]
    check_supported: Optional[str]
    payload_information: Optional[str]
    description: Optional[str]
    refs: Optional[str]
    type: Optional[str]
    payload_default: Optional[str]
    target: Optional[str]
    session_required: Optional[str]
    option_headings: List["ModOptGroup"] = Relationship(back_populates="vector")
    targets: List["Aim"] = Relationship(back_populates="vector")
    payloads: List["Artifact"] = Relationship(back_populates="vectors", link_model=VectorArtifactLink)

# models/links.py
from sqlmodel import SQLModel, Field
from typing import Optional
class VectorArtifactLink(SQLModel, table=True):
    __tablename__ = "attacks_attack_payload"
    __table_args__ = {"extend_existing": True}
    attack_id: Optional[int] = Field(default=None, foreign_key="attacks_attack.attack_id", primary_key=True)
    payload_id: Optional[int] = Field(default=None, foreign_key="attacks_payload.payload_id", primary_key=True)

# models/opts.py
from typing import List, Optional, TYPE_CHECKING
from sqlmodel import SQLModel, Field, Relationship
if TYPE_CHECKING:
    from .vector import Vector
    from .artifact import Artifact
class ModOptGroup(SQLModel, table=True):
    __tablename__ = "attacks_option_heading"
    option_heading_id: Optional[int] = Field(default=None, primary_key=True)
    attack_id: int = Field(foreign_key="attacks_attack.attack_id")
    title: Optional[str]
    name: Optional[str]
    type: Optional[str]
    order_by: Optional[str]
    vector: "Vector" = Relationship(back_populates="option_headings")
    module_options: List["ModOpt"] = Relationship(back_populates="option_group")
class ModOpt(SQLModel, table=True):
    __tablename__ = "attacks_option"
    option_id: Optional[int] = Field(default=None, primary_key=True)
    name: Optional[str]
    current_setting: Optional[str]
    required: Optional[str]
    description: Optional[str]
    order_by: Optional[str]
    option_heading_id: int = Field(foreign_key="attacks_option_heading.option_heading_id")
    option_group: ModOptGroup = Relationship(back_populates="module_options")
class ArtifactOptGroup(SQLModel, table=True):
    __tablename__ = "attacks_payload_option_heading"
    payload_option_heading_id: Optional[int] = Field(default=None, primary_key=True)
    payload_id: int = Field(foreign_key="attacks_payload.payload_id")
    title: Optional[str]
    name: Optional[str]
    type: Optional[str]
    order_by: Optional[str]
    artifact: "Artifact" = Relationship(back_populates="payload_headings")
    payload_options: List["ArtifactOpt"] = Relationship(back_populates="payload_option_group")
class ArtifactOpt(SQLModel, table=True):
    __tablename__ = "attacks_payload_option"
    payload_option_id: Optional[int] = Field(default=None, primary_key=True)
    name: Optional[str]
    current_setting: Optional[str]
    required: Optional[str]
    description: Optional[str]
    order_by: Optional[str]
    payload_option_heading_id: int = Field(foreign_key="attacks_payload_option_heading.payload_option_heading_id")
    payload_option_group: ArtifactOptGroup = Relationship(back_populates="payload_options")

# models/artifact.py
from typing import List, Optional, TYPE_CHECKING
from sqlmodel import SQLModel, Field, Relationship
from .links import VectorArtifactLink
if TYPE_CHECKING:
    from .vector import Vector
    from .opts import ArtifactOptGroup
class Artifact(SQLModel, table=True):
    __tablename__ = "attacks_payload"
    payload_id: Optional[int] = Field(default=None, primary_key=True)
    order_by: Optional[str]
    payload: Optional[str]
    disclosure: Optional[str]
    rank: Optional[str]
    description: Optional[str]
    check_supported: Optional[str]
    payload_headings: List["ArtifactOptGroup"] = Relationship(back_populates="artifact")
    vectors: List["Vector"] = Relationship(back_populates="payloads", link_model=VectorArtifactLink)

# models/aim.py
from typing import Optional
from sqlmodel import SQLModel, Field, Relationship
from .vector import Vector
class Aim(SQLModel, table=True):
    __tablename__ = "attacks_target"
    target_id: Optional[int] = Field(default=None, primary_key=True)
    id: Optional[str]
    name: Optional[str]
    default_setting: Optional[str]
    order_by: Optional[str]
    attack_id: Optional[int] = Field(default=None, foreign_key="attacks_attack.attack_id")
    vector: Optional[Vector] = Relationship(back_populates="targets")

# models/api_schemas.py
from typing import List, Optional
from pydantic import BaseModel
class PayloadOptionResponse(BaseModel):
    id: Optional[str]
    order_by: Optional[str]
    name: Optional[str]
    current_setting: Optional[str]
    required: Optional[str]
    description: Optional[str]
class PayloadOptionHeadingResponse(BaseModel):
    payload_id: Optional[int]
    order_by: Optional[str]
    name: Optional[str]
    title: Optional[str]
    type: Optional[str]
    payload_options: List[PayloadOptionResponse] = []
class PayloadResponse(BaseModel):
    payload_id: Optional[int]
    order_by: Optional[str]
    payload: Optional[str]
    disclosure: Optional[str]
    rank: Optional[str]
    description: Optional[str]
    check_supported: Optional[str]
    payload_headings: List[PayloadOptionHeadingResponse] = []
class ModuleOptionResponse(BaseModel):
    id: Optional[str]
    order_by: Optional[str]
    name: Optional[str]
    current_setting: Optional[str]
    required: Optional[str]
    description: Optional[str]
class ModuleOptionHeadingResponse(BaseModel):
    attack_id: Optional[int]
    order_by: Optional[int]
    name: Optional[str]
    title: Optional[str]
    type: Optional[str]
    module_options: List[ModuleOptionResponse] = []
class TargetResponse(BaseModel):
    target_id: Optional[int]
    id: Optional[int]
    name: Optional[str]
    default_setting: Optional[str]
    order_by: Optional[int]
class VectorResponse(BaseModel):
    attack_id: int
    name: Optional[str]
    module: Optional[str]
    platform: Optional[str]
    arch: Optional[str]
    privileged: Optional[str]
    license: Optional[str]
    rank: Optional[str]
    disclosed: Optional[str]
    provided_by: Optional[str]
    module_side_effects: Optional[str]
    module_stability: Optional[str]
    module_reliability: Optional[str]
    check_supported: Optional[str]
    payload_information: Optional[str]
    description: Optional[str]
    refs: Optional[str]
    type: Optional[str]
    payload_default: Optional[str]
    option_headings: List[ModuleOptionHeadingResponse] = []
    payload_options: List[PayloadResponse] = []
    targets: List[TargetResponse] = []
class VectorBrief(BaseModel):
    attack_id: Optional[int]
    name: str
    module: str
    rank: Optional[str]
    disclosed: Optional[str]
    session_required: Optional[str]
    type: Optional[str]
    refs: Optional[str]
    description: Optional[str]
class TargetResponse(BaseModel):
    target_id: int
    id: int
    name: str
    default_setting: str
    order_by: int

# api.py
from typing import List
from fastapi import Depends, FastAPI, HTTPException
from sqlmodel import Session, select
from models.vector import Vector
from models.api_schemas import VectorBrief
app = FastAPI()
# Assume get_session is defined elsewhere and returns a SQLModel Session
def list_vectors(
    session: Session = Depends(get_session),
    offset: int = 0,
    limit: int = 100,
) -> List[VectorBrief]:
    return session.exec(
        select(
            Vector.attack_id,
            Vector.name,
            Vector.module,
            Vector.rank,
            Vector.disclosed,
            Vector.session_required,
            Vector.type,
            Vector.refs,
            Vector.description,
        )
        .offset(offset)
        .limit(limit)
    ).all()
app.get("/attacks", response_model=List[VectorBrief])(list_vectors)
@app.post("/attacks")
def collect_vectors(
    attack_ids: List[int],
    session: Session = Depends(get_session),
):
    out = []
    for vec_id in attack_ids:
        vec = session.get(Vector, vec_id)
        if not vec:
            continue
        out.append(compose_vector(vec))
    return out
@app.get("/attacks/{attack_id}")
def get_vector(attack_id: int, session: Session = Depends(get_session)):
    vec = session.get(Vector, attack_id)
    if not vec:
        raise HTTPException(status_code=404, detail="Attack not found")
    return compose_vector(vec)
def compose_vector(vec: Vector):
    attack_id = vec.attack_id
    payload_options = []
    for art in vec.payloads:
        for hdr in art.payload_headings:
            hdr_opts = []
            for opt in hdr.payload_options:
                hdr_opts.append({
                    "option_name": opt.name,
                    "option_value": opt.current_setting,
                    "option_required": opt.required,
                    "option_description": opt.description,
                    "option_order_by": opt.order_by,
                })
            payload_options.append({
                "payload_id": hdr.payload_id,
                "payload_name": hdr.payload.payload,
                "payload_order_by": hdr.order_by,
                "payload_options": hdr_opts,
            })
    option_headings = []
    for grp in vec.option_headings:
        group_opts = []
        if "Payload" not in grp.title:
            for opt in grp.module_options:
                group_opts.append({
                    "option_name": opt.name,
                    "option_value": opt.current_setting,
                    "option_required": opt.required,
                    "option_description": opt.description,
                    "option_order_by": opt.order_by,
                })
            option_headings.append({
                "module_name": grp.name,
                "module_title": grp.title,
                "module_order_by": grp.order_by,
                "module_options": group_opts,
            })
    targets = []
    for t in vec.targets:
        targets.append({
            "target_id": t.target_id,
            "id": t.id,
            "name": t.name,
            "default_setting": t.default_setting,
            "order_by": t.order_by,
        })
    return {
        "attack_id": attack_id,
        "module": vec.module,
        "name": vec.name,
        "platform": vec.platform,
        "arch": vec.arch,
        "privileged": vec.privileged,
        "license": vec.license,
        "rank": vec.rank,
        "disclosed": vec.disclosed,
        "provided_by": vec.provided_by,
        "module_side_effects": vec.module_side_effects,
        "module_stability": vec.module_stability,
        "module_reliability": vec.module_reliability,
        "check_supported": vec.check_supported,
        "payload_information": vec.payload_information,
        "description": vec.description,
        "refs": vec.refs,
        "type": vec.type,
        "payload_default": vec.payload_default,
        "payload_options": payload_options,
        "module_options": option_headings,
        "target_options": targets,
        "target": vec.target,
        "session_required": vec.session_required,
    }

What changed and why it works

The crucial shift is returning a nested object that mirrors the intended JSON, instead of returning a list of tuples from a join. The ORM relationships make navigation intuitive, but they do not automatically convert flat rows into a tree. The function that assembles the response walks related entities and produces dictionaries with stable keys. The response models define the shape for documentation and validation, and selects in the list endpoint keep the listing payload lean.

Why this matters

APIs that expose joined rows leak persistence concerns into the boundary and force consumers to reverse-engineer meaning from flat structures. Once the schema grows to include groups of options, payload settings, and targets, raw rows become unmanageable and validation errors compound. A response model that reflects how clients actually consume data makes documentation clear, validation predictable, and day-two maintenance far less painful.

Takeaways

If a join returns the right data but FastAPI’s validator complains, look at the response model first. When you need deep nesting, prefer explicit relationships on table models and build the JSON shape intentionally, as above. That keeps your contract stable, your OpenAPI readable, and your endpoints resilient as the schema grows.