Release v2

nyekiapi/api/__init__.py

@@ -0,0 +1,22 @@
+from fastapi import FastAPI
+from fastapi.responses import ORJSONResponse
+
+from nyekiapi.lib import settings
+from nyekiapi.api.lifespan import lifespan
+from nyekiapi.api.v2 import app as app_v2
+from nyekiapi.api.v2.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("/v2", app_v2, name="v2")

nyekiapi/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,
+            }
+        )

nyekiapi/api/lifespan.py

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

nyekiapi/api/v2/__init__.py

@@ -0,0 +1,35 @@
+from fastapi import FastAPI
+from fastapi.responses import ORJSONResponse
+from fastapi.exceptions import RequestValidationError
+
+from nyekiapi.lib import settings
+from nyekiapi.api import docs
+from nyekiapi.api.v2.errors import (
+    BaseError,
+    NotFoundError,
+    ValidationError,
+    InternalServerError,
+)
+from nyekiapi.api.v2.profiles.router import router as example_router
+
+
+app = FastAPI(
+    debug=settings.DEBUG,
+    title="Nyeki's API",
+    version="2.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)
+app.add_exception_handler(RequestValidationError, ValidationError.handler)
+app.add_exception_handler(500, InternalServerError.handler)
+app.add_exception_handler(Exception, InternalServerError.handler)
+
+
+docs.load_onto_fastapi("docs/v2", app)

nyekiapi/api/v2/errors.py

@@ -0,0 +1,100 @@
+from fastapi.responses import ORJSONResponse
+
+from nyekiapi.api.v2.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 if hasattr(exc, "status_code") else cls.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 BadGatewayError(ErrorInitMixin, BaseError):
+    title: str = "Bad Gateway"
+    message: str = "The server received an invalid response from the upstream server."
+    status_code: int = 502
+
+
+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

nyekiapi/api/v2/profiles/router.py

@@ -0,0 +1,158 @@
+from enum import StrEnum
+
+from typing import Annotated, Self
+
+from httpx import HTTPStatusError
+
+from fastapi import APIRouter, Query, Response
+
+from pydantic import BaseModel, Field, model_validator
+
+from nyekiapi.lib import settings
+from nyekiapi.api.v2.errors import NotFoundError, BadGatewayError, InternalServerError
+from nyekiapi.api.v2.schemas import ErrorSchema
+from nyekiapi.api.v2.profiles.schemas import ProfileSchema
+from nyekiapi.lib.integrations import discord, github, Profile
+
+
+router = APIRouter(tags=["Profiles"])
+
+
+sources = []
+
+if settings.DISCORD_TOKEN:
+    sources.append(("DISCORD", "discord"))
+
+if settings.GITHUB_TOKEN:
+    sources.append(("GITHUB", "github"))
+
+
+ProfileSource = StrEnum("ProfileSource", sources)
+
+
+class ProfileQuerySchema(BaseModel):
+    source: ProfileSource = Field(..., description="The source of the profile.")
+    source_id: str | None = Field(
+        None,
+        description='The ID of the profile in the source.\nSupported by: `"discord"` `"github"`',
+    )
+    source_name: str | None = Field(
+        None,
+        description='The name of the profile in the source.\nSupported by: `"github"`',
+    )
+
+    @model_validator(mode="after")
+    def validate_source(self) -> Self:
+        if self.source_id and self.source_name:
+            raise ValueError(
+                "Both source_id and source_name cannot be provided at the same time."
+            )
+
+        if self.source.value == "discord" and not self.source_id:
+            raise ValueError("source_id is required for Discord profiles.")
+
+        if self.source.value == "github" and not (self.source_name or self.source_id):
+            raise ValueError(
+                "Either source_name or source_id is required for GitHub profiles."
+            )
+
+        return self
+
+
+ProfileQuery = Annotated[ProfileQuerySchema, Query()]
+
+
+async def _get_profile(profile: ProfileQuerySchema) -> Profile:
+    try:
+        if profile.source.value == "discord":
+            return await discord.get_profile_by_id(profile.source_id)
+
+        elif profile.source.value == "github":
+            if profile.source_id:
+                return await github.get_profile_by_id(profile.source_id)
+            else:
+                return await github.get_profile_by_name(profile.source_name)
+
+    except HTTPStatusError as e:
+        if e.response.status_code == 404:
+            raise NotFoundError("profile") from e
+
+        elif e.response.status_code >= 500:
+            raise BadGatewayError() from e
+
+        else:
+            raise InternalServerError() from e
+
+
+@router.get(
+    "/profile",
+    responses={
+        404: {"model": ErrorSchema},
+        422: {"model": ErrorSchema},
+        500: {"model": ErrorSchema},
+        502: {"model": ErrorSchema},
+    },
+)
+async def get_profile(profile: ProfileQuery) -> ProfileSchema:
+    """Returns basic information about a profile from a given source."""
+
+    profile = await _get_profile(profile)
+
+    return ProfileSchema.from_integration(profile)
+
+
+@router.get(
+    "/profile/avatar",
+    responses={
+        404: {"model": ErrorSchema},
+        422: {"model": ErrorSchema},
+        500: {"model": ErrorSchema},
+        502: {"model": ErrorSchema},
+    },
+    status_code=302,
+)
+async def get_profile_avatar(response: Response, profile: ProfileQuery) -> None:
+    """Redirects to the avatar URL of a profile from a given source."""
+
+    profile = await _get_profile(profile)
+
+    response.headers["Location"] = profile.avatar_url
+
+
+@router.get(
+    "/profile/banner",
+    responses={
+        404: {"model": ErrorSchema},
+        422: {"model": ErrorSchema},
+        500: {"model": ErrorSchema},
+        502: {"model": ErrorSchema},
+    },
+    status_code=302,
+)
+async def get_profile_banner(response: Response, profile: ProfileQuery) -> None:
+    """Redirects to the banner URL of a profile from a given source."""
+
+    profile = await _get_profile(profile)
+
+    if not profile.banner_url:
+        raise NotFoundError("profile banner")
+
+    response.headers["Location"] = profile.banner_url
+
+
+@router.get(
+    "/profile/url",
+    responses={
+        404: {"model": ErrorSchema},
+        422: {"model": ErrorSchema},
+        500: {"model": ErrorSchema},
+        502: {"model": ErrorSchema},
+    },
+    status_code=302,
+)
+async def get_profile_url(response: Response, profile: ProfileQuery) -> None:
+    """Redirects to the profile URL of a profile from a given source."""
+
+    profile = await _get_profile(profile)
+
+    response.headers["Location"] = profile.url

nyekiapi/api/v2/profiles/schemas.py

@@ -0,0 +1,29 @@
+from pydantic import BaseModel, Field
+
+from nyekiapi.lib.integrations import Profile
+
+
+class ProfileSchema(BaseModel):
+    id: str = Field(
+        ..., description="The unique identifier of the profile at the source."
+    )
+    url: str = Field(..., description="The URL of the profile at the source.")
+    name: str = Field(..., description="The display name of the profile at the source.")
+    username: str = Field(..., description="The username of the profile at the source.")
+    avatar_url: str = Field(
+        ..., description="The avatar URL of the profile at the source."
+    )
+    banner_url: str | None = Field(
+        None, description="The banner URL of the profile at the source."
+    )
+
+    @classmethod
+    def from_integration(cls, profile: Profile) -> "ProfileSchema":
+        return cls(
+            id=profile.id,
+            url=profile.url,
+            name=profile.name,
+            username=profile.username,
+            avatar_url=profile.avatar_url,
+            banner_url=profile.banner_url,
+        )

nyekiapi/api/v2/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 = Field(..., description="A short, human-readable title of the error.")
+    message: str = Field(..., description="A more detailed description of the error.")