WIP: Litestar

Signed-off-by: Edgar Ramírez Mondragón <edgarrm358@gmail.com>

Author: renovate[bot]

pyproject.toml

@@ -11,6 +11,7 @@ dependencies = [
     "aiosqlite~=0.22.0",
     "fastapi==0.128.3",
     "granian[reload]~=2.7.0",
+    "litestar[standard]~=2.20.0",
     "packaging~=26.0",
     "pydantic~=2.12",
     "python-json-logger~=4.0",
@@ -56,6 +57,9 @@ exclude-newer = "P7D"
 preview = true
 required-version = ">=0.9.17"
 
+# [tool.uv.sources]
+# litestar = { git = "https://github.com/litestar-org/litestar", rev = "main" }
+
 [tool.deptry.per_rule_ignores]
 DEP002 = [
     "granian",  # Not imported, but used as the ASGI server as a uvicorn alternative
@@ -114,6 +118,8 @@ addopts = ["--durations=5", "-ra", "--strict-config", "--strict-markers"]
 asyncio_default_fixture_loop_scope = "session"
 filterwarnings = [
     "error",
+    "once:Core Pydantic V1 functionality isn't compatible with Python 3.14 or greater:UserWarning",
+    "once:Call to deprecated function 'resolve_before_request'. Deprecated in litestar 3.0. This function will be removed in 4.0. Use '.before_request attribute' instead:DeprecationWarning",
 ]
 log_cli_level = "INFO"
 minversion = "9"

src/hub_api/client.py

@@ -388,6 +388,7 @@ async def get_plugin_index(self: MeltanoHub) -> api_schemas.PluginIndex:
                 plugins[plugin_type][plugin_name] = api_schemas.PluginRef(
                     default_variant=default_variant,
                     logo_url=logo_http_url,
+                    variants={},
                 )
 
             plugins[plugin_type][plugin_name].variants[variant_name] = api_schemas.VariantReference(
@@ -435,6 +436,7 @@ async def get_plugin_type_index(
                 plugins[plugin_name] = api_schemas.PluginRef(
                     default_variant=default_variant,
                     logo_url=logo_http_url,
+                    variants={},
                 )
 
             plugins[plugin_name].variants[variant_name] = api_schemas.VariantReference(
@@ -448,6 +450,38 @@ async def get_plugin_type_index(
 
         return plugins
 
+    async def get_plugin_variants(self: MeltanoHub, plugin_id: ids.PluginID) -> api_schemas.PluginRef:
+        """Get plugin variants."""
+        sql = """
+            SELECT
+                p.name AS plugin,
+                p.plugin_type,
+                pv.logo_url,
+                pv.name AS variant
+            FROM plugins p
+            JOIN plugin_variants pv ON pv.plugin_id = p.id
+            WHERE p.id = :plugin_id AND p.plugin_type = :plugin_type
+        """
+        ref = api_schemas.PluginRef(default_variant="", logo_url=None, variants={})
+        params = {
+            "plugin_id": plugin_id.as_db_id(),
+            "plugin_type": plugin_id.plugin_type.value,
+        }
+        for row in await fetch_all_dicts(self.db, sql, params):
+            if not ref.default_variant:
+                ref.default_variant = row["variant"]
+                # breakpoint()
+                ref.logo_url = pydantic.HttpUrl(f"{self.base_hub_url}{row['logo_url']}") if row["logo_url"] else None
+            ref.variants[row["variant"]] = api_schemas.VariantReference(
+                ref=_build_variant_path(
+                    plugin_type=plugin_id.plugin_type,
+                    plugin_name=plugin_id.plugin_name,
+                    plugin_variant=row["variant"],
+                    base_url=self.base_url,
+                ),
+            )
+        return ref
+
     async def get_sdk_plugins(
         self: MeltanoHub,
         *,

src/hub_api/litestar_app/app.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+import http
+
+from litestar import Litestar, MediaType, Request, Response
+from litestar.config.compression import CompressionConfig
+from litestar.exceptions import MethodNotAllowedException
+from litestar.plugins.pydantic import PydanticPlugin
+
+from hub_api.exceptions import NotFoundError
+
+from . import maintainers, plugins
+
+
+def not_found_exception_handler(_: Request, exc: NotFoundError) -> Response:  # type: ignore[type-arg]
+    """Default handler for exceptions subclassed from HTTPException."""
+    return Response(
+        media_type=MediaType.JSON,
+        content={"details": exc.args[0]},
+        status_code=http.HTTPStatus.NOT_FOUND,
+    )
+
+
+def method_not_allowed_exception_handler(_: Request, exc: MethodNotAllowedException) -> Response:  # type: ignore[type-arg]
+    """Default handler for exceptions subclassed from HTTPException."""
+    return Response(
+        media_type=MediaType.JSON,
+        content={"details": exc.args[0]},
+        status_code=http.HTTPStatus.METHOD_NOT_ALLOWED,
+        headers=exc.headers,
+    )
+
+
+app = Litestar(
+    [
+        maintainers.MaintainersController,
+        plugins.PluginsController,
+    ],
+    compression_config=CompressionConfig(backend="gzip", minimum_size=1000),
+    exception_handlers={  # ty: ignore[invalid-argument-type]
+        NotFoundError: not_found_exception_handler,
+        MethodNotAllowedException: method_not_allowed_exception_handler,
+    },
+    plugins=[
+        PydanticPlugin(
+            exclude_none=True,
+            exclude_unset=True,
+        ),
+    ],
+    debug=True,
+)

src/hub_api/litestar_app/dependencies.py

@@ -0,0 +1,21 @@
+"""Litestar app dependencies."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from hub_api import client, database
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncGenerator
+
+    from litestar.connection import Request
+
+
+async def get_hub(request: Request) -> AsyncGenerator[client.MeltanoHub]:  # type: ignore[type-arg]
+    """Get a Meltano hub instance."""
+    db = await database.open_db()
+    try:
+        yield client.MeltanoHub(db=db, base_url=str(request.base_url))
+    finally:
+        await db.close()

src/hub_api/litestar_app/maintainers.py

@@ -0,0 +1,68 @@
+"""Maintainer endpoints."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Annotated, ClassVar
+
+from litestar import Controller, get
+from litestar.di import Provide
+from litestar.openapi.spec.example import Example
+from litestar.params import Parameter
+
+from hub_api import client  # noqa: TC001
+from hub_api.schemas import api as api_schemas  # noqa: TC001
+
+from .dependencies import get_hub
+
+if TYPE_CHECKING:
+    from litestar.types.composite_types import Dependencies
+
+
+class MaintainersController(Controller):
+    path = "/meltano/api/v1/maintainers"
+
+    dependencies: ClassVar[Dependencies] = {  # type: ignore[misc]
+        "hub": Provide(get_hub),
+    }
+
+    @get("/", summary="Get maintainers list")
+    async def get_maintainers(self, hub: client.MeltanoHub) -> api_schemas.MaintainersList:  # noqa: PLR6301
+        """Retrieve global index of plugins."""
+        return await hub.get_maintainers()
+
+    @get("/top", summary="Get top plugin maintainers")
+    async def get_top_maintainers(  # noqa: PLR6301
+        self,
+        hub: client.MeltanoHub,
+        count: Annotated[
+            int,
+            Parameter(
+                ...,
+                ge=1,
+                lt=50,
+                description="The number of maintainers to return",
+            ),
+        ] = 10,
+    ) -> list[api_schemas.MaintainerPluginCount]:
+        """Retrieve top maintainers."""
+        return await hub.get_top_maintainers(count)
+
+    @get("/{maintainer:str}", summary="Get maintainer details")
+    async def get_maintainer(  # noqa: PLR6301
+        self,
+        hub: client.MeltanoHub,
+        maintainer: Annotated[
+            str,
+            Parameter(
+                ...,
+                description="The maintainer identifier",
+                pattern=r"^[A-Za-z0-9-_]+$",
+                examples=[
+                    Example(value="meltanolabs"),
+                    Example(value="singer-io"),
+                ],
+            ),
+        ],
+    ) -> api_schemas.MaintainerDetails:
+        """Retrieve maintainer details."""
+        return await hub.get_maintainer(maintainer)