Initial commit

project/__init__.py

@@ -0,0 +1 @@
+from project.api import app as app

project/api/__init__.py

@@ -0,0 +1,22 @@
+from fastapi import FastAPI
+from fastapi.responses import ORJSONResponse
+
+from project.lib import settings
+from project.api.lifespan import lifespan
+from project.api.v1 import app as app_v1
+from project.api.v1.errors import NotFoundError
+
+
+app = FastAPI(
+    debug=settings.DEBUG,
+    docs_url=None,
+    redoc_url=None,
+    lifespan=lifespan,
+    default_response_class=ORJSONResponse,
+)
+
+
+app.add_exception_handler(404, NotFoundError.handler)
+
+
+app.mount("/v1", app_v1, name="v1")

project/api/docs.py

@@ -0,0 +1,137 @@
+import os
+
+from enum import Enum
+
+from dataclasses import dataclass
+
+from fastapi import FastAPI
+from fastapi.routing import APIRoute
+
+
+@dataclass
+class OpenAPI:
+    summary: str | None
+    description: str | None
+    tags: list["OpenAPITag"]
+
+    @classmethod
+    def load(cls, base_path: str) -> "OpenAPI":
+        """Loads the OpenAPI documentation strings from a path.
+
+        Inside that path:
+            _Summary.md -> The OpenAPI summary.
+            _Description.md -> The OpenAPI description.
+            *.md -> Tags, the file name without the prefix is the tag name and the file
+                contents is the description.
+
+        Args:
+            base_path (str): The path to the directory where the files are.
+
+        Returns:
+            OpenAPI: The initialized OpenAPI object.
+        """
+
+        summary = None
+        description = None
+        tags = []
+
+        dirs = os.listdir(base_path)
+        dirs.sort()
+
+        for doc_file in dirs:
+            if not doc_file.endswith(".md"):
+                continue
+
+            with open(f"{base_path}/{doc_file}", "r", encoding="utf-8") as f:
+                content = f.read()
+
+                name = doc_file.split(".")[0]
+
+                if name == "_Summary":
+                    summary = content
+
+                elif name == "_Description":
+                    description = content
+
+                else:
+                    tags.append(OpenAPITag(name=name, description=content))
+
+        return cls(summary=summary, description=description, tags=tags)
+
+
+@dataclass
+class OpenAPITag:
+    name: str
+    description: str
+
+
+def load(base_path: str) -> OpenAPI:
+    """Loads the OpenAPI documentation strings from a path.
+
+    Inside that path:
+        _Summary.md -> The OpenAPI summary.
+        _Description.md -> The OpenAPI description.
+        *.md -> Tags, the file name without the prefix is the tag name and the file
+            contents is the description.
+
+    Args:
+        base_path (str): The path to the directory where the files are.
+
+    Returns:
+        OpenAPI: The initialized OpenAPI object.
+    """
+
+    return OpenAPI.load(base_path)
+
+
+def load_onto_fastapi(
+    base_path: str,
+    app: FastAPI,
+    *,
+    only_tags: list[str | Enum] = None,
+    only_used_tags: bool = True,
+) -> None:
+    """Loads the OpenAPI summaries and descriptions and loads them onto a `FastAPI`
+    application.
+
+    Read `load()`'s documentation for information about structure.
+
+    Args:
+        base_path (str): The path to the directory where the files are.
+        app (FastAPI): The app to load the docs onto.
+        only_tags (list[str | Enum], optional): A list of tags to load. If not
+            provided, all tags will be loaded. Defaults to None.
+        only_used_tags (bool, optional): If True, only tags that are used in the
+            app's routes will be loaded. Defaults to True.
+    """
+
+    openapi = OpenAPI.load(base_path)
+
+    if only_tags is not None and only_used_tags:
+        raise ValueError("Cannot use both `only_tags` and `only_used_tags`.")
+
+    if only_used_tags:
+        used_tags = set()
+
+        for route in app.routes:
+            if isinstance(route, APIRoute):
+                for tag in route.tags:
+                    used_tags.add(tag)
+
+        openapi.tags = [tag for tag in openapi.tags if tag.name in used_tags]
+
+    elif only_tags is not None:
+        only_tags = [tag.value if isinstance(tag, Enum) else tag for tag in only_tags]
+        openapi.tags = [tag for tag in openapi.tags if tag.name in only_tags]
+
+    app.summary = openapi.summary
+    app.description = openapi.description
+    app.openapi_tags = []
+
+    for tag in openapi.tags:
+        app.openapi_tags.append(
+            {
+                "name": tag.name,
+                "description": tag.description,
+            }
+        )

project/api/lifespan.py

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

project/api/v1/__init__.py

@@ -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)

project/api/v1/errors.py

@@ -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

project/api/v1/example/router.py

@@ -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!")

project/api/v1/example/schemas.py

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

project/api/v1/schemas.py

@@ -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 = Field(..., description="A short, human-readable title of the error.")
+    message: str = Field(..., description="A more detailed description of the error.")
+
+
+class Page[T](BaseModel):
+    """A standard paginated response schema."""
+
+    items: list[T] = Field(..., description="The items in the current page.")
+    total: int = Field(..., description="The amount of items available across all pages for this query.")
+
+    @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.
+"""

project/db/__init__.py

@@ -0,0 +1 @@
+from project.db.models.example import Book as Book

project/db/fields.py

@@ -0,0 +1,26 @@
+import nanoid
+
+from tortoise import fields
+
+
+def NanoIDField(*args, **kwargs):
+    """A nanoid DB field.
+
+    This is an alias for a CharField with a max length of 21 characters, using
+    `nanoid.generate` as the default value generator. The arguments and keyword
+    arguments are passed directly to the CharField constructor.
+
+    Arguments:
+        *args: Positional arguments for the CharField.
+        **kwargs: Keyword arguments for the CharField.
+
+    Returns:
+        tortoise.fields.CharField: A CharField configured as a nanoid primary key.
+    """
+
+    return fields.CharField(
+        max_length=21,
+        default=nanoid.generate,
+        *args,
+        **kwargs,
+    )

project/db/lifespan.py

@@ -0,0 +1,17 @@
+from tortoise import Tortoise
+from tortoise.migrations import api as migrations
+
+from project.lib import settings
+
+
+async def on_startup():
+    """Connects to the database and creates the missing schemas on FastAPI startup."""
+
+    await migrations.migrate(config=settings.DATABASE)
+    await Tortoise.init(config=settings.DATABASE, _enable_global_fallback=True)
+
+
+async def on_shutdown():
+    """Closes the database connections on FastAPI shutdown."""
+
+    await Tortoise.close_connections()

project/db/models/example.py

@@ -0,0 +1,16 @@
+from tortoise import fields
+from tortoise.models import Model
+
+from project.db.fields import NanoIDField
+
+
+class Book(Model):
+    id = NanoIDField(primary_key=True)
+
+    title = fields.CharField(max_length=255)
+    author = fields.CharField(max_length=255)
+
+    published_at = fields.DatetimeField(null=True)
+
+    created_at = fields.DatetimeField(auto_now_add=True)
+    updated_at = fields.DatetimeField(auto_now=True)

project/lib/settings.py

@@ -0,0 +1,25 @@
+import os
+
+from dotenv import load_dotenv
+
+
+load_dotenv()
+
+
+DEBUG = os.getenv("DEBUG", "False").lower() in ("true", "1", "t")
+
+
+DATABASE = {
+    "connections": {"default": os.getenv("DATABASE_URL", "sqlite://db.sqlite3")},
+    "apps": {
+        "models": {
+            "models": [
+                "project.db.models.example",
+            ],
+            "default_connection": "default",
+            "migrations": "project.db.migrations",
+        }
+    },
+    "use_tz": True,
+    "timezone": "UTC",
+}