🎨 Make Model.Schema callable with readonly arg

This commit is contained in:
Brian Wiborg 2025-09-29 18:22:43 +02:00
parent 29a5018ae3
commit c56ea6451e
No known key found for this signature in database
5 changed files with 29 additions and 22 deletions

View file

@ -125,6 +125,8 @@ Next, create your endpoints in `tournament/routes.py`:
from ohmyapi.router import APIRouter, HTTPException, HTTPStatus from ohmyapi.router import APIRouter, HTTPException, HTTPStatus
from ohmyapi.db.exceptions import DoesNotExist from ohmyapi.db.exceptions import DoesNotExist
from typing import List
from .models import Tournament from .models import Tournament
# OhMyAPI will automatically pick up all instances of `fastapi.APIRouter` and # OhMyAPI will automatically pick up all instances of `fastapi.APIRouter` and
@ -137,23 +139,23 @@ from .models import Tournament
tournament_router = APIRouter(prefix="/tournament", tags=['Tournament']) tournament_router = APIRouter(prefix="/tournament", tags=['Tournament'])
@tournament_router.get("/") @tournament_router.get("/", response_model=List[Tournament.Schema()])
async def list(): async def list():
queryset = Tournament.all() queryset = Tournament.all()
return await Tournament.Schema.model.from_queryset(queryset) return await Tournament.Schema.model.from_queryset(queryset)
@tournament_router.post("/", status_code=HTTPStatus.CREATED) @tournament_router.post("/", status_code=HTTPStatus.CREATED)
async def post(tournament: Tournament.Schema.readonly): async def post(tournament: Tournament.Schema(readonly=True)):
queryset = Tournament.create(**payload.model_dump()) queryset = Tournament.create(**payload.model_dump())
return await Tournament.Schema.model.from_queryset(queryset) return await Tournament.Schema().from_queryset(queryset)
@tournament_router.get("/:id") @tournament_router.get("/:id", response_model=Tournament.Schema())
async def get(id: str): async def get(id: str):
try: try:
queryset = Tournament.get(id=id) queryset = Tournament.get(id=id)
return await Tournament.Schema.model.from_queryset_single(tournament) return await Tournament.Schema().from_queryset_single(tournament)
except DoesNotExist: except DoesNotExist:
raise HTTPException(status_code=404, detail="not found") raise HTTPException(status_code=404, detail="not found")
@ -273,7 +275,7 @@ router = APIRouter(prefix="/tournament", tags=["Tournament"])
@router.get("/") @router.get("/")
async def list(user: auth.User = Depends(permissions.require_authenticated)): async def list(user: auth.User = Depends(permissions.require_authenticated)):
queryset = Tournament.all() queryset = Tournament.all()
return await Tournament.Schema.model.from_queryset(queryset) return await Tournament.Schema().from_queryset(queryset)
... ...
@ -315,7 +317,7 @@ router = APIRouter(prefix="/tournament", tags=["Tournament"])
@router.get("/teams") @router.get("/teams")
async def teams(user: auth.User = Depends(permissions.require_authenticated)): async def teams(user: auth.User = Depends(permissions.require_authenticated)):
queryset = Team.for_user(user) queryset = Team.for_user(user)
return await Tournament.Schema.model.from_queryset(queryset) return await Tournament.Schema().from_queryset(queryset)
``` ```
## Shell ## Shell

View file

@ -207,7 +207,7 @@ async def introspect(token: Dict = Depends(get_token)):
return token return token
@router.get("/me", response_model=User.Schema.model) @router.get("/me", response_model=User.Schema())
async def me(user: User = Depends(get_current_user)): async def me(user: User = Depends(get_current_user)):
"""Return the currently authenticated user.""" """Return the currently authenticated user."""
return await User.Schema.model.from_tortoise_orm(user) return await User.Schema().from_tortoise_orm(user)

View file

@ -8,29 +8,29 @@ from typing import List
# Expose your app's routes via `router = fastapi.APIRouter`. # Expose your app's routes via `router = fastapi.APIRouter`.
# Use prefixes wisely to avoid cross-app namespace-collisions. # Use prefixes wisely to avoid cross-app namespace-collisions.
# Tags improve the UX of the OpenAPI docs at /docs. # Tags improve the UX of the OpenAPI docs at /docs.
router = APIRouter(prefix="/tournemant") router = APIRouter(prefix="/tournament")
@router.get( @router.get(
"/", tags=["tournament"], response_model=List[models.Tournament.Schema.model] "/", tags=["tournament"], response_model=List[models.Tournament.Schema()]
) )
async def list(): async def list():
"""List all tournaments.""" """List all tournaments."""
return await models.Tournament.Schema.model.from_queryset(models.Tournament.all()) return await models.Tournament.Schema().from_queryset(models.Tournament.all())
@router.post("/", tags=["tournament"], status_code=HTTPStatus.CREATED) @router.post("/", tags=["tournament"], status_code=HTTPStatus.CREATED)
async def post(tournament: models.Tournament.Schema.readonly): async def post(tournament: models.Tournament.Schema(readonly=True)):
"""Create tournament.""" """Create tournament."""
return await models.Tournament.Schema.model.from_queryset( return await models.Tournament.Schema().from_queryset(
models.Tournament.create(**tournament.model_dump()) models.Tournament.create(**tournament.model_dump())
) )
@router.get("/{id}", tags=["tournament"], response_model=models.Tournament.Schema.model) @router.get("/{id}", tags=["tournament"], response_model=models.Tournament.Schema())
async def get(id: str): async def get(id: str):
"""Get tournament by id.""" """Get tournament by id."""
return await models.Tournament.Schema.model.from_queryset( return await models.Tournament.Schema().from_queryset(
models.Tournament.get(id=id) models.Tournament.get(id=id)
) )
@ -43,12 +43,12 @@ async def get(id: str):
) )
async def put(tournament: models.Tournament.Schema.model): async def put(tournament: models.Tournament.Schema.model):
"""Update tournament.""" """Update tournament."""
return await models.Tournament.Schema.model.from_queryset( return await models.Tournament.Schema().from_queryset(
models.Tournament.update(**tournament.model_dump()) models.Tournament.update(**tournament.model_dump())
) )
@router.delete("/{id}", tags=["tournament"]) @router.delete("/{id}", status_code=HTTPStatus.ACCEPTED, tags=["tournament"])
async def delete(id: str): async def delete(id: str):
try: try:
tournament = await models.Tournament.get(id=id) tournament = await models.Tournament.get(id=id)

View file

@ -14,16 +14,16 @@ from typing import List
router = APIRouter(prefix="/{{ app_name }}", tags=['{{ app_name }}']) router = APIRouter(prefix="/{{ app_name }}", tags=['{{ app_name }}'])
@router.get("/") @router.get("/", response_model=List)
async def list(): async def list():
"""List all ...""" """List all ..."""
return [] return []
@router.post("/") @router.post("/", status_code=HTTPStatus.CREATED)
async def post(): async def post():
"""Create ...""" """Create ..."""
return HTTPException(status_code=HTTPStatus.CREATED) raise HTTPException(status_code=HTTPStatus.IM_A_TEAPOT)
@router.get("/{id}") @router.get("/{id}")
@ -38,7 +38,7 @@ async def put(id: str):
return HTTPException(status_code=HTTPStatus.ACCEPTED) return HTTPException(status_code=HTTPStatus.ACCEPTED)
@router.delete("/{id}") @router.delete("/{id}", status_code=HTTPStatus.ACCEPTED)
async def delete(id: str): async def delete(id: str):
return HTTPException(status_code=HTTPStatus.ACCEPTED) return HTTPException(status_code=HTTPStatus.ACCEPTED)

View file

@ -36,6 +36,11 @@ class ModelMeta(type(TortoiseModel)):
schema_opts = getattr(new_cls, "Schema", None) schema_opts = getattr(new_cls, "Schema", None)
class BoundSchema: class BoundSchema:
def __call__(self, readonly: bool = False):
if readonly:
return self.readonly
return self.model
@property @property
def model(self): def model(self):
"""Return a Pydantic model class for serializing results.""" """Return a Pydantic model class for serializing results."""