[Python] How do I write a join in FastAPI with SQLModel?

almost exact question, 2 years old, no responses: Deep nesting response model in fastapi using sqlmodel

My question is like this one but I have an error: Not able to get data for joining two tables in fastapi using sqlmodel. I am so close, I just cannot parse the output.

so the data looks like:

[
{
attack_id: 123,
attack_name: "meta exploit",

options: [
{ option_id: 444, option_name: "svhost" }
]
}
]


and I am new to fastapi. I want to return the data from a get request like that. Or like anything, really, I just need a really fast api. I found out how to do a join and the data looks like it is being returned correctly based on the ERROR. The error is hard to read (tons of fields in "real" data).

this code does work for just returning the Attack table.

from typing import Annotated

import logging

from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select

logger = logging.getLogger('uvicorn.error')

class Attack(SQLModel, table=True):
__tablename__ = "attacks_attack"

attack_id: int | None = Field(default=None, primary_key=True)
name: str
module: str

class Option(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 AttackOption(SQLModel):
attack_id: int | None = Field(default=None, primary_key=True)
name: str
module: str
options: tuple[Option]


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 create_db_and_tables():
SQLModel.metadata.create_all(engine)


def get_session():
with Session(engine) as session:
yield session


SessionDep = Annotated[Session, Depends(get_session)]

app = FastAPI()


@app.get("/attacks/")
def read_heroes(
session: SessionDep,
offset: int = 0,
limit: Annotated[int, Query(le=100)] = 1,
) -> list[AttackOption]:
heroes = session.exec(select(Attack, Option).join(Option).offset(offset).limit(limit)).all()
logger.info(heroes)
return heroes


and my error:

raise ResponseValidationError(
fastapi.exceptions.ResponseValidationError: 18 validation errors:
{'type': 'missing', 'loc': ('response', 0, 'name'), 'msg': 'Field required', 'input': (Attack(name='invscout RPM Privilege Escalation', attack_id=1, module='exploit/aix/local/invscout_rpm_priv_esc', platform='Unix, AIX', privileged='No', ... (503 characters truncated) ... d.nist.gov/vuln/detail/CVE-2023-28528\nhttp://web.archive.org/web/20250117163943/https://talosintelligence.com/vulnerability_reports/TALOS-2023-1691'), Option(required='yes', option_id=1, order_by='1', name='INVSCOUT_PATH', current_setting='/usr/sbin/invscout', description='Path to invscout executable', attack_id=1))}
{'type': 'missing', 'loc': ('response', 0, 'module'), 'msg': 'Field required', 'input': (Attack(name='invscout RPM Privilege Escalation', attack_id=1, module='exploit/aix/local/invscout_rpm_priv_esc', platform='Unix, AIX', privileged='No', ... (503 characters truncated) ... d.nist.gov/vuln/detail/CVE-2023-28528\nhttp://web.archive.org/web/20250117163943/https://talosintelligence.com/vulnerability_reports/TALOS-2023-1691'), Option(required='yes', option_id=1, order_by='1', name='INVSCOUT_PATH', current_setting='/usr/sbin/invscout', description='Path to invscout executable', attack_id=1))}


currently investigating: Getting nested (joined) tables to display in the OpenAPI interface provided by FastAPI and SQLModel as the answer. I need to apply it though

I got this to happen:

[(Attack(name='invscout RPM Privilege Escalation', attack_id=1, module='exploit/aix/local/invscout_rpm_priv_esc', platform='Unix, AIX', privileged='No', ... (503 characters truncated) ... d.nist.gov/vuln/detail/CVE-2023-28528\nhttp://web.archive.org/web/20250117163943/https://talosintelligence.com/vulnerability_reports/TALOS-2023-1691'), Option(required='yes', option_id=1, order_by='1', name='INVSCOUT_PATH', current_setting='/usr/sbin/invscout', description='Path to invscout executable', attack_id=1)), (Attack(name='invscout RPM Privilege Escalation', attack_id=1, module='exploit/aix/local/invscout_rpm_priv_esc', platform='Unix, AIX', privileged='No', ... (503 characters truncated) ... d.nist.gov/vuln/detail/CVE-2023-28528\nhttp://web.archive.org/web/20250117163943/https://talosintelligence.com/vulnerability_reports/TALOS-2023-1691'), Option(required='yes', option_id=2, order_by='2', name='SESSION', current_setting='', description='The session to run this module on', attack_id=1))]


with:

@app.get("/attacks/")
def read_heroes(
session: SessionDep,
offset: int = 0,
limit: Annotated[int, Query(le=100)] = 2,
) -> list[AttackOption]:
heroes = session.exec(select(Attack, Option).join(Option).offset(offset).limit(limit)).all()
logger.info(heroes)
return heroes


class AttackOption(SQLModel):
attack_id: int | None = Field(default=None, primary_key=True)
options: list["Option"] = Relationship(back_populates="Attack")


even though the json response is essentially two rows of null. but that is not exactly what I need. It is two rows for one option.

one validation error now:
fastapi.exceptions.ResponseValidationError: 1 validation errors:
{'type': 'model_attributes_type', 'loc': ('response',), 'msg': 'Input should be a valid dictionary or object to extract fields from', 'input': [(Attack(type='exploits', arch='cmd', license='Metasploit Framework License (BSD)', disclosed='2023-04-24', module_side_effects='ioc-in-logs', module_re ... (503 characters truncated) ... 'Yes', description='This module exploits a command injection vulnerability in IBM AIX\ninvscout set-uid root utility present in AIX 7.2 and earlier.'), Option(name='INVSCOUT_PATH', current_setting='/usr/sbin/invscout', description='Path to invscout executable', attack_id=1, option_id=1, required='yes', order_by='1')), (Attack(type='exploits', arch='cmd', license='Metasploit Framework License (BSD)', disclosed='2023-04-24', module_side_effects='ioc-in-logs', module_re ... (503 characters truncated) ... 'Yes', description='This module exploits a command injection vulnerability in IBM AIX\ninvscout set-uid root utility present in AIX 7.2 and earlier.'), Option(name='SESSION', current_setting='', description='The session to run this module on', attack_id=1, option_id=2, required='yes', order_by='2'))]}

@app.get("/customers/{customer_id}", status_code=200, response_model=Option)
def read_customer(customer_id: int, db: Session = Depends(get_session),
offset: int = 0,
limit: Annotated[int, Query(le=100)] = 2,):
heroes = db.exec(select(Attack, Option).join(Option).offset(offset).limit(limit)).all()
if heroes is None:
raise HTTPException(status_code=404, detail=f"Customer not found for {customer_id=}")
return heroes

Continue reading...