WIP: Litestar

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

Author: renovate[bot]

pyproject.toml

@@ -12,6 +12,7 @@ dependencies = [
     "aiosqlite==0.22.1",
     "fastapi==0.139.2",
     "granian[reload]==2.8.0",
+    "litestar[standard]==2.24.0",
     "packaging==26.2",
     "pydantic==2.14.0a1",
     "python-json-logger==4.1.0",

src/hub_api/client.py

@@ -389,6 +389,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(
@@ -436,6 +437,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(
@@ -449,6 +451,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,63 @@
+"""Maintainer endpoints."""
+
+from __future__ import annotations
+
+from typing import Annotated
+
+from litestar import Controller, get
+from litestar.di import NamedDependency, Provide
+from litestar.openapi.spec.example import Example
+from litestar.params import PathParameter, QueryParameter
+
+from hub_api import client  # ruff: ignore[typing-only-first-party-import]
+from hub_api.schemas import api as api_schemas  # ruff: ignore[typing-only-first-party-import]
+
+from .dependencies import get_hub
+
+
+class MaintainersController(Controller):
+    path = "/meltano/api/v1/maintainers"
+
+    dependencies = {  # ruff: ignore[mutable-class-default]
+        "hub": Provide(get_hub),
+    }
+
+    @get("/", summary="Get maintainers list")
+    async def get_maintainers(self, hub: NamedDependency[client.MeltanoHub]) -> api_schemas.MaintainersList:  # ruff: ignore[no-self-use]
+        """Retrieve global index of plugins."""
+        return await hub.get_maintainers()
+
+    @get("/top", summary="Get top plugin maintainers")
+    async def get_top_maintainers(  # ruff: ignore[no-self-use]
+        self,
+        hub: NamedDependency[client.MeltanoHub],
+        count: Annotated[
+            int,
+            QueryParameter(
+                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(  # ruff: ignore[no-self-use]
+        self,
+        hub: NamedDependency[client.MeltanoHub],
+        maintainer: Annotated[
+            str,
+            PathParameter(
+                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)