First version

project/api/__init__.py

@@ -0,0 +1,27 @@
+from fastapi import FastAPI
+from fastapi.responses import ORJSONResponse
+
+from project.lib import settings
+from project.api import docs
+from project.api.errors import BaseError, NotFoundError
+from project.api.lifespan import lifespan
+from project.api.example.router import router as example_router
+
+
+app = FastAPI(
+    debug=settings.DEBUG,
+    title="Project API",
+    description=docs.OPENAPI_DESCRIPTION,
+    version="1.0.0",
+    docs_url=None,
+    redoc_url="/docs",
+    lifespan=lifespan,
+    openapi_tags=docs.OPENAPI_TAGS,
+    default_response_class=ORJSONResponse,
+)
+
+
+app.include_router(example_router)
+
+app.add_exception_handler(BaseError, BaseError.handler)
+app.add_exception_handler(404, NotFoundError.handler)

project/api/docs.py

@@ -0,0 +1,27 @@
+import os
+
+
+OPENAPI_TAGS = []
+
+OPENAPI_DESCRIPTION = ""
+
+
+for doc_file in os.listdir("./docs"):
+    if not doc_file.endswith(".md"):
+        continue
+
+    with open(f"./docs/{doc_file}", "r", encoding="utf-8") as f:
+        content = f.read()
+
+        name = doc_file.split(".")[0]
+
+        if name == "_Project":
+            OPENAPI_DESCRIPTION = content
+
+        else:
+            OPENAPI_TAGS.append(
+                {
+                    "name": name,
+                    "description": content,
+                }
+            )

project/api/errors.py

@@ -0,0 +1,94 @@
+from fastapi.responses import ORJSONResponse
+
+from project.api.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

project/api/example/router.py

@@ -0,0 +1,13 @@
+from fastapi import APIRouter
+
+from project.api.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!")

project/api/example/schemas.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel
+
+
+class MessageSchema(BaseModel):
+    message: str

project/api/lifespan.py

@@ -0,0 +1,14 @@
+from fastapi import FastAPI
+
+from contextlib import asynccontextmanager
+
+from project.db.lifespan import on_startup, on_shutdown
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    await on_startup()
+
+    yield
+
+    await on_shutdown()

project/api/schemas.py

@@ -0,0 +1,29 @@
+from typing import Annotated
+
+from pydantic import BaseModel, Field
+
+
+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