gistfile1.txt

from typing import List, Iterator
from contextlib import asynccontextmanager
from pydantic import BaseModel
from sqlmodel import SQLModel, Session, Field, Relationship, create_engine, select
from fastapi import FastAPI, Depends


class Kind(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    pets: List["Pet"] = Relationship(back_populates="kind")


class Pet(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(max_length=20)
    kind_id: int | None = Field(foreign_key="kind.id")
    kind: Kind = Relationship(back_populates="pets")


class PetRead(BaseModel):
    id: int
    name: str
    kind: Kind


class KindRead(BaseModel):
    id: int
    name: str
    pets: List[Pet] | None


connect_args = {"check_same_thread": False}
engine = create_engine("sqlite:///test.db", echo=True, connect_args=connect_args)


def create_db_and_tables() -> None:
    SQLModel.metadata.create_all(engine)


def get_session() -> Iterator[Session]:
    with Session(engine) as session:
        yield session


@asynccontextmanager
async def life_span(app2: FastAPI):
    create_db_and_tables()
    yield


app = FastAPI(lifespan=life_span)


@app.get('/')
def hello():
    return {'detail': 'Hello world'}


@app.get('/pet')
def pet(session: Session = Depends(get_session)) -> List[PetRead]:
    return session.exec(select(Pet)).all()


@app.get('/pet/{pet_id}')
def pet_by_id(pet_id: int, session: Session = Depends(get_session)) -> PetRead:
    return session.get(Pet, pet_id)


@app.get('/kind/{kind_id}')
def kind_by_id(kind_id: int, session: Session = Depends(get_session)) -> KindRead:
    return session.get(Kind, kind_id)