TL;DR: How do I give FastAPI and OpenAPI the information about Relationships between Entities? --------------------------Kontext-------------------------- So I recently started creating a python Backend Server with FastAPI that I wanted to test, while developing it. While trying out various approaches I found schemathesis and it worked really well for the simple stuff. But then I wanted it to handle relationships between entities properly. With that I mean stuff like customer -n:m-> addresses and similar. But that doesn't seem to work well, because I don't know if and how to generate an openapi doc that schemathesis can use to handle those relationships properly: These are the DTOs used to create the schemas for openapi. class CustomerCreate(BaseModel): username: str firstname: str lastname: str address_id: str email: Optional[str] = Field(..., json_schema_extra={ "pattern": EMAIL_REGEX.pattern, "examples": ["user@example.com"]}) phone: Optional[str] = "" customer_addresses: Optional[List[CustomerAddressCreate]] = None @field_validator("email") @classmethod def email_validators(cls, v): v = validate_email(v) return v class CustomerAddressCreate(BaseModel): customer_id: str address_id: str address_type: str class AddressCreate(BaseModel): street: str city: str region: str | None = None postal_code: str | None = None country: str subregion: str | None = None building: str | None = None locality: str | None = None address_line2: str | None = None address_line3: str | None = None latitude: float | None = None longitude: float | None = None Obviously these DTOs, don't contain information about Foreign Keys an such, therefore all schemathesis tests fail because they try to generate entities with empty ids and such. So the question is: --------------------------Question-------------------------- How do I give FastAPI and OpenAPI the information about Relationships? My objective is that schemathesis is getting the information about the relationships and is able to e.g. create a customer, and address and a finally a customer-address based on existing ids of customer and address. Bonus points if I don't have to create my own test cases since I am using schemathesis mostly from the cli. In my mind having to test entities that have relationships with each other should be a common use-case but I ----------------Appendix (info that might help):---------------- The general Setup is: Python Backend with FastAPI Documentation with in built OpenAPI Database: PostgreSQL Api Tests: Schemathesis using pure CLI no Authentication yet. will be added later. A bit of tests with pytest General goal for now: Add a lot more Entities and have schemathesis test the new ones along the way. I don't have the time and resources to test every single endpoint that is supposed to work the same anyways. All Endpoints are based on this Generic Entity_API: # entity_api.py from typing import Generic, Type, TypeVar, Optional from fastapi import APIRouter, Depends, Query from pydantic import BaseModel from sqlalchemy.orm import Session from .entity_service import EntityService from db.session import SessionLocal M = TypeVar("M") CreateT = TypeVar("CreateT", bound=BaseModel) UpdateT = TypeVar("UpdateT", bound=BaseModel) ReadT = TypeVar("ReadT", bound=BaseModel) IDT = TypeVar("IDT", int, str) def get_db(): db = SessionLocal() try: yield db finally: db.close() class EntityAPI(Generic[M, CreateT, UpdateT, ReadT, IDT]): # ---- subclass must set these ---- model: Type[M] Create: Type[CreateT] Update: Type[UpdateT] Read: Type[ReadT] ID: Optional[Type[IDT]] = None prefix: str = "" tags: tuple[str, ...] = () service: EntityService[M, CreateT, UpdateT, ReadT, IDT] # ---- optional knobs (subclasses may override) ---- # Add OpenAPI extras per route (e.g., links under responses) # e.g. {201: {"links": {...}}} create_responses: Optional[dict] = None read_responses: Optional[dict] = None search_responses: Optional[dict] = None update_responses: Optional[dict] = None delete_responses: Optional[dict] = None # Control status codes if you want 201 on POST, etc. create_status_code: int = 201 update_status_code: int = 200 read_status_code: int = 200 search_status_code: int = 200 delete_status_code: int = 204 # Stable operationIds (auto if not overridden) op_create: Optional[str] = None op_read: Optional[str] = None op_search: Optional[str] = None op_update: Optional[str] = None op_delete: Optional[str] = None def __init__(self) -> None: if not hasattr(self, "service"): self.service = EntityService( model=self.model, read_schema=self.Read) self.router = APIRouter(prefix=self.prefix, tags=list(self.tags)) # Bind concrete types so wrappers capture them Create = self.Create Update = self.Update Read = self.Read ID = self.ID # Derive default operationIds from the prefix if not provided base = self.prefix.strip( "/").replace("/", "_") or self.model.__name__.lower() op_create = self.op_create or f"{base}_create" op_read = self.op_read or f"{base}_read" op_search = self.op_search or f"{base}_search" op_update = self.op_update or f"{base}_update" op_delete = self.op_delete or f"{base}_delete" # ---- Per-instance wrapper functions ---- def _route_create(item, db: Session = Depends(get_db)): return self.create(item, db) _route_create.__annotations__ = {"item": Create, "return": Read} def _route_read(item_id, db: Session = Depends(get_db)): return self.read(item_id, db) _route_read.__annotations__ = {"item_id": ID, "return": Read} def _route_search(q: Optional[str] = Query(None), db: Session = Depends(get_db)): return self.search(q, db) _route_search.__annotations__ = {"return": list[Read]} def _route_update(item_id, item, db: Session = Depends(get_db)): return self.update(item_id, item, db) _route_update.__annotations__ = { "item_id": ID, "item": Update, "return": Read} def _route_delete(item_id, db: Session = Depends(get_db)): return self.delete(item_id, db) # ---- Register routes (now with operation_id & responses) ---- self.router.post( "/", response_model=Read, status_code=self.create_status_code, operation_id=op_create, responses=self.create_responses or {}, )(_route_create) self.router.get( "/{item_id}", response_model=Read, status_code=self.read_status_code, operation_id=op_read, responses=self.read_responses or {}, )(_route_read) self.router.get( "/", response_model=list[Read], status_code=self.search_status_code, operation_id=op_search, responses=self.search_responses or {}, )(_route_search) self.router.put( "/{item_id}", response_model=Read, status_code=self.update_status_code, operation_id=op_update, responses=self.update_responses or {}, )(_route_update) self.router.delete( "/{item_id}", status_code=self.delete_status_code, operation_id=op_delete, responses=self.delete_responses or {}, )(_route_delete) # ---------- default CRUD logic ---------- def create(self, item: CreateT, db: Session): return self.service.create(item, db) def read(self, item_id: IDT, db: Session): return self.service.read(item_id, db) def search(self, q: Optional[str], db: Session): return self.service.search(q, db) def update(self, item_id: IDT, item: UpdateT, db: Session): return self.service.update(item_id, item, db) def delete(self, item_id: IDT, db: Session): return self.service.delete(item_id, db) And a singular Entity can look like this: from fastapi import APIRouter from entity.entity_api import EntityAPI from entity.entity_service import EntityService from .customer_model import Customer from .customer_dto import CustomerCreate, CustomerRead, CustomerUpdate class CustomerAPI(EntityAPI[Customer, CustomerCreate, CustomerUpdate, CustomerRead, str]): model = Customer Create = CustomerCreate Update = CustomerUpdate Read = CustomerRead op_create = "customers_create" ID = str prefix = "/customers" tags = ("customers",) service = EntityService(model=Customer, read_schema=CustomerRead) # Export the router for app.py router: APIRouter = CustomerAPI().router Continue reading...
