Template
mirror of
https://github.com/Nekidev/uv-fastapi-tortoise.git
synced 2026-09-12 19:57:25 +00:00
Add versioning support by default, replace _Project.md with _Summary.md and _Description.md, upgrade to the latest Tortoise ORM version
This commit is contained in:
@@ -0,0 +1,26 @@
|
||||
from fastapi import FastAPI
|
||||
from fastapi.responses import ORJSONResponse
|
||||
|
||||
from project.lib import settings
|
||||
from project.api import docs
|
||||
from project.api.v1.errors import BaseError, NotFoundError
|
||||
from project.api.v1.example.router import router as example_router
|
||||
|
||||
|
||||
app = FastAPI(
|
||||
debug=settings.DEBUG,
|
||||
title="Project API",
|
||||
version="1.0.0",
|
||||
docs_url=None,
|
||||
redoc_url="/docs",
|
||||
default_response_class=ORJSONResponse,
|
||||
)
|
||||
|
||||
|
||||
app.include_router(example_router)
|
||||
|
||||
app.add_exception_handler(BaseError, BaseError.handler)
|
||||
app.add_exception_handler(404, NotFoundError.handler)
|
||||
|
||||
|
||||
docs.load_onto_fastapi("docs/v1", app)
|
||||
@@ -0,0 +1,94 @@
|
||||
from fastapi.responses import ORJSONResponse
|
||||
|
||||
from project.api.v1.schemas import ErrorSchema
|
||||
|
||||
|
||||
class BaseError(Exception):
|
||||
status_code: int = 500
|
||||
title: str = "Error"
|
||||
message: str = "An error occurred."
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
title: str = ...,
|
||||
message: str = ...,
|
||||
status_code: int = ...,
|
||||
):
|
||||
if title is not ...:
|
||||
self.title = title
|
||||
|
||||
if message is not ...:
|
||||
self.message = message
|
||||
|
||||
if status_code is not ...:
|
||||
self.status_code = status_code
|
||||
|
||||
@classmethod
|
||||
async def handler(cls, _request, exc: "BaseError"):
|
||||
"""Handles the exception and returns a standardized error response."""
|
||||
|
||||
return ORJSONResponse(
|
||||
status_code=exc.status_code,
|
||||
content=ErrorSchema(
|
||||
title=exc.title if hasattr(exc, "title") else cls.title,
|
||||
message=exc.message if hasattr(exc, "message") else cls.message,
|
||||
).model_dump(mode="json"),
|
||||
)
|
||||
|
||||
def __repr__(self):
|
||||
return f"{self.__class__.__name__}(title={self.title!r}, message={self.message!r}, status_code={self.status_code!r})"
|
||||
|
||||
|
||||
class ErrorInitMixin:
|
||||
def __init__(self, message: str = ..., *, title: str = ...):
|
||||
if title is not ...:
|
||||
self.title = title
|
||||
|
||||
if message is not ...:
|
||||
self.message = message
|
||||
|
||||
|
||||
class InternalServerError(ErrorInitMixin, BaseError):
|
||||
title: str = "Internal Server Error"
|
||||
message: str = "An unexpected error occurred on the server."
|
||||
status_code: int = 500
|
||||
|
||||
|
||||
class ValidationError(ErrorInitMixin, BaseError):
|
||||
title: str = "Validation Error"
|
||||
message: str = "One or more validation errors occurred."
|
||||
status_code: int = 422
|
||||
|
||||
|
||||
class ConflictError(ErrorInitMixin, BaseError):
|
||||
title: str = "Conflict"
|
||||
message: str = "The request could not be completed due to a conflict with the current state of the resource."
|
||||
status_code: int = 409
|
||||
|
||||
|
||||
class NotFoundError(BaseError):
|
||||
title: str = "Not Found"
|
||||
message: str = "The requested resource was not found."
|
||||
status_code: int = 404
|
||||
|
||||
def __init__(self, resource: str = "resource"):
|
||||
self.message = f"The requested {resource} was not found."
|
||||
|
||||
|
||||
class ForbiddenError(ErrorInitMixin, BaseError):
|
||||
title: str = "Forbidden"
|
||||
message: str = "You do not have permission to access this resource."
|
||||
status_code: int = 403
|
||||
|
||||
|
||||
class PaymentRequiredError(ErrorInitMixin, BaseError):
|
||||
title: str = "Payment Required"
|
||||
message: str = "Payment is required to access this resource."
|
||||
status_code: int = 402
|
||||
|
||||
|
||||
class UnauthorizedError(ErrorInitMixin, BaseError):
|
||||
title: str = "Unauthorized"
|
||||
message: str = "You must be authenticated to access this resource."
|
||||
status_code: int = 401
|
||||
@@ -0,0 +1,13 @@
|
||||
from fastapi import APIRouter
|
||||
|
||||
from project.api.v1.example.schemas import MessageSchema
|
||||
|
||||
|
||||
router = APIRouter(tags=["Example"])
|
||||
|
||||
|
||||
@router.get("/hello")
|
||||
async def hello_world() -> MessageSchema:
|
||||
"""An example API endpoint that returns a hello world message."""
|
||||
|
||||
return MessageSchema(message="Hello, World!")
|
||||
@@ -0,0 +1,5 @@
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class MessageSchema(BaseModel):
|
||||
message: str
|
||||
@@ -0,0 +1,75 @@
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import Query
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from tortoise.queryset import QuerySet
|
||||
|
||||
|
||||
NanoID = Annotated[
|
||||
str, Field(min_length=21, max_length=21, pattern="^[A-Za-z0-9_-]{21}$")
|
||||
]
|
||||
"""A nanoid identifier type for pydantic models.
|
||||
|
||||
Example:
|
||||
```
|
||||
from pydantic import BaseModel
|
||||
|
||||
from project.api.schemas import NanoID
|
||||
|
||||
|
||||
class BookSchema(BaseModel):
|
||||
id: NanoID
|
||||
title: str
|
||||
```
|
||||
"""
|
||||
|
||||
|
||||
class ErrorSchema(BaseModel):
|
||||
"""A standard error response schema."""
|
||||
|
||||
title: str
|
||||
message: str
|
||||
|
||||
|
||||
class Page[T](BaseModel):
|
||||
"""A standard paginated response schema."""
|
||||
|
||||
items: list[T]
|
||||
total: int
|
||||
|
||||
@classmethod
|
||||
async def from_queryset(
|
||||
cls, schema: type[BaseModel], qs: QuerySet, params: "PageParams"
|
||||
) -> "Page[T]":
|
||||
"""Creates a paginated response from a Tortoise ORM QuerySet.
|
||||
|
||||
Args:
|
||||
schema (type[Schema]): The schema type to serialize items with.
|
||||
qs (QuerySet): The Tortoise ORM QuerySet to paginate.
|
||||
params (PageParams): The pagination parameters.
|
||||
|
||||
Returns:
|
||||
Page[T]: A paginated response containing serialized items.
|
||||
"""
|
||||
|
||||
total = await qs.count()
|
||||
items = await qs.limit(params.limit).offset(params.offset)
|
||||
|
||||
return cls(items=[schema.from_orm(item) for item in items], total=total)
|
||||
|
||||
|
||||
class _PageParams(BaseModel):
|
||||
"""The standard pagination query parameters."""
|
||||
|
||||
limit: int = Field(25, le=100)
|
||||
offset: int = 0
|
||||
|
||||
|
||||
PageParams = Annotated[_PageParams, Query()]
|
||||
"""The standard pagination query parameters annotation.
|
||||
|
||||
It wraps a `_PageParams` model for use in FastAPI route definitions without requiring
|
||||
`Annotated[PageParams, Query()]` on every use.
|
||||
"""
|
||||
Reference in New Issue
Block a user