Merge pull request #27 from modern-python/22-feature-add-swagger-instrument

implement swagger instrument

Author: lesnik512

lite_bootstrap/bootstrappers/fastapi_bootstrapper.py

@@ -3,6 +3,7 @@
 
 from lite_bootstrap import import_checker
 from lite_bootstrap.bootstrappers.base import BaseBootstrapper
+from lite_bootstrap.fastapi_offline_docs.main import enable_offline_docs
 from lite_bootstrap.instruments.cors_instrument import CorsConfig, CorsInstrument
 from lite_bootstrap.instruments.healthchecks_instrument import (
     HealthChecksConfig,
@@ -13,6 +14,7 @@
 from lite_bootstrap.instruments.opentelemetry_instrument import OpentelemetryConfig, OpenTelemetryInstrument
 from lite_bootstrap.instruments.prometheus_instrument import PrometheusConfig, PrometheusInstrument
 from lite_bootstrap.instruments.sentry_instrument import SentryConfig, SentryInstrument
+from lite_bootstrap.instruments.swagger_instrument import SwaggerConfig, SwaggerInstrument
 
 
 if import_checker.is_fastapi_installed:
@@ -28,7 +30,9 @@
 
 
 @dataclasses.dataclass(kw_only=True, slots=True, frozen=True)
-class FastAPIConfig(CorsConfig, HealthChecksConfig, LoggingConfig, OpentelemetryConfig, PrometheusConfig, SentryConfig):
+class FastAPIConfig(
+    CorsConfig, HealthChecksConfig, LoggingConfig, OpentelemetryConfig, PrometheusConfig, SentryConfig, SwaggerConfig
+):
     application: "fastapi.FastAPI" = dataclasses.field(default_factory=lambda: fastapi.FastAPI())
     opentelemetry_excluded_urls: list[str] = dataclasses.field(default_factory=list)
     prometheus_instrumentator_params: dict[str, typing.Any] = dataclasses.field(default_factory=dict)
@@ -129,6 +133,18 @@ def bootstrap(self) -> None:
         )
 
 
+@dataclasses.dataclass(kw_only=True, frozen=True)
+class FastApiSwaggerInstrument(SwaggerInstrument):
+    bootstrap_config: FastAPIConfig
+
+    def bootstrap(self) -> None:
+        self.bootstrap_config.application.docs_url = self.bootstrap_config.swagger_path
+        if self.bootstrap_config.swagger_offline_docs:
+            enable_offline_docs(
+                self.bootstrap_config.application, static_files_handler=self.bootstrap_config.service_static_path
+            )
+
+
 class FastAPIBootstrapper(BaseBootstrapper["fastapi.FastAPI"]):
     __slots__ = "bootstrap_config", "instruments"
 
@@ -139,6 +155,7 @@ class FastAPIBootstrapper(BaseBootstrapper["fastapi.FastAPI"]):
         FastAPIHealthChecksInstrument,
         FastAPILoggingInstrument,
         FastAPIPrometheusInstrument,
+        FastApiSwaggerInstrument,
     ]
     bootstrap_config: FastAPIConfig
     not_ready_message = "fastapi is not installed"

lite_bootstrap/bootstrappers/litestar_bootstrapper.py

@@ -1,4 +1,5 @@
 import dataclasses
+import pathlib
 import typing
 
 from lite_bootstrap import import_checker
@@ -18,22 +19,32 @@
     PrometheusInstrument,
 )
 from lite_bootstrap.instruments.sentry_instrument import SentryConfig, SentryInstrument
+from lite_bootstrap.instruments.swagger_instrument import SwaggerConfig, SwaggerInstrument
 
 
 if import_checker.is_litestar_installed:
     import litestar
     from litestar.config.app import AppConfig
     from litestar.config.cors import CORSConfig
     from litestar.contrib.opentelemetry import OpenTelemetryConfig
+    from litestar.openapi import OpenAPIConfig
+    from litestar.openapi.plugins import SwaggerRenderPlugin
     from litestar.plugins.prometheus import PrometheusConfig, PrometheusController
+    from litestar.static_files import create_static_files_router
 
 if import_checker.is_opentelemetry_installed:
     from opentelemetry.trace import get_tracer_provider
 
 
 @dataclasses.dataclass(kw_only=True, slots=True, frozen=True)
 class LitestarConfig(
-    CorsConfig, HealthChecksConfig, LoggingConfig, OpentelemetryConfig, PrometheusBootstrapperConfig, SentryConfig
+    CorsConfig,
+    HealthChecksConfig,
+    LoggingConfig,
+    OpentelemetryConfig,
+    PrometheusBootstrapperConfig,
+    SentryConfig,
+    SwaggerConfig,
 ):
     application_config: "AppConfig" = dataclasses.field(default_factory=lambda: AppConfig())
     opentelemetry_excluded_urls: list[str] = dataclasses.field(default_factory=list)
@@ -130,6 +141,41 @@ class LitestarPrometheusController(PrometheusController):
         self.bootstrap_config.application_config.middleware.append(litestar_prometheus_config.middleware)
 
 
+@dataclasses.dataclass(kw_only=True, frozen=True)
+class LitestarSwaggerInstrument(SwaggerInstrument):
+    bootstrap_config: LitestarConfig
+
+    def bootstrap(self) -> None:
+        render_plugins: typing.Final = (
+            (
+                SwaggerRenderPlugin(
+                    js_url=f"{self.bootstrap_config.service_static_path}/swagger-ui-bundle.js",
+                    css_url=f"{self.bootstrap_config.service_static_path}/swagger-ui.css",
+                    standalone_preset_js_url=(
+                        f"{self.bootstrap_config.service_static_path}/swagger-ui-standalone-preset.js"
+                    ),
+                ),
+            )
+            if self.bootstrap_config.swagger_offline_docs
+            else (SwaggerRenderPlugin(),)
+        )
+        self.bootstrap_config.application_config.openapi_config = OpenAPIConfig(
+            path=self.bootstrap_config.swagger_path,
+            title=self.bootstrap_config.service_name,
+            version=self.bootstrap_config.service_version,
+            description=self.bootstrap_config.service_description,
+            render_plugins=render_plugins,
+            **self.bootstrap_config.swagger_extra_params,
+        )
+        if self.bootstrap_config.swagger_offline_docs:
+            static_dir_path = pathlib.Path(__file__).parent.parent / "litestar_swagger_static"
+            self.bootstrap_config.application_config.route_handlers.append(
+                create_static_files_router(
+                    path=self.bootstrap_config.service_static_path, directories=[static_dir_path]
+                )
+            )
+
+
 class LitestarBootstrapper(BaseBootstrapper["litestar.Litestar"]):
     __slots__ = "bootstrap_config", "instruments"
 
@@ -140,6 +186,7 @@ class LitestarBootstrapper(BaseBootstrapper["litestar.Litestar"]):
         LitestarHealthChecksInstrument,
         LitestarLoggingInstrument,
         LitestarPrometheusInstrument,
+        LitestarSwaggerInstrument,
     ]
     bootstrap_config: LitestarConfig
     not_ready_message = "litestar is not installed"

lite_bootstrap/fastapi_offline_docs/__init__.py

@@ -0,0 +1,61 @@
+import os
+import pathlib
+import typing
+
+from lite_bootstrap import import_checker
+
+
+if import_checker.is_fastapi_installed:
+    from fastapi import FastAPI, Request
+    from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html, get_swagger_ui_oauth2_redirect_html
+    from fastapi.responses import HTMLResponse
+    from fastapi.staticfiles import StaticFiles
+    from starlette.routing import Route
+
+
+def enable_offline_docs(
+    app: "FastAPI",
+    static_files_handler: str = "/static",
+    static_dir_path: os.PathLike[str] = pathlib.Path(__file__).parent / "static",
+    include_docs_in_schema: bool = False,
+) -> None:
+    if not (app_openapi_url := app.openapi_url):
+        msg = "No app.openapi_url specified"
+        raise RuntimeError(msg)
+
+    docs_url: str = app.docs_url or "/docs"
+    redoc_url: str = app.redoc_url or "/redoc"
+    swagger_ui_oauth2_redirect_url: str = app.swagger_ui_oauth2_redirect_url or "/docs/oauth2-redirect"
+
+    app.router.routes = [
+        route
+        for route in app.router.routes
+        if typing.cast(Route, route).path not in (docs_url, redoc_url, swagger_ui_oauth2_redirect_url)
+    ]
+
+    app.mount(static_files_handler, StaticFiles(directory=static_dir_path), name=static_files_handler)
+
+    @app.get(docs_url, include_in_schema=include_docs_in_schema)
+    async def custom_swagger_ui_html(request: Request) -> HTMLResponse:
+        root_path = typing.cast(str, request.scope.get("root_path", "").rstrip("/"))
+        swagger_js_url = f"{root_path}{static_files_handler}/swagger-ui-bundle.js"
+        swagger_css_url = f"{root_path}{static_files_handler}/swagger-ui.css"
+        return get_swagger_ui_html(
+            openapi_url=root_path + app_openapi_url,
+            title=f"{app.title} - Swagger UI",
+            oauth2_redirect_url=app.swagger_ui_oauth2_redirect_url,
+            swagger_js_url=swagger_js_url,
+            swagger_css_url=swagger_css_url,
+        )
+
+    @app.get(swagger_ui_oauth2_redirect_url, include_in_schema=include_docs_in_schema)
+    async def swagger_ui_redirect() -> HTMLResponse:
+        return get_swagger_ui_oauth2_redirect_html()
+
+    @app.get(redoc_url, include_in_schema=include_docs_in_schema)
+    async def redoc_html() -> HTMLResponse:
+        return get_redoc_html(
+            openapi_url=app_openapi_url,
+            title=f"{app.title} - ReDoc",
+            redoc_js_url=f"{static_files_handler}/redoc.standalone.js",
+        )

lite_bootstrap/fastapi_offline_docs/main.py

@@ -0,0 +1,9 @@
+import re
+import typing
+
+
+VALID_PATH_PATTERN: typing.Final = re.compile(r"^(/[a-zA-Z0-9_-]+)+/?$")
+
+
+def is_valid_path(maybe_path: str) -> bool:
+    return bool(re.fullmatch(VALID_PATH_PATTERN, maybe_path))

lite_bootstrap/fastapi_offline_docs/static/redoc.standalone.js

@@ -5,6 +5,7 @@
 @dataclasses.dataclass(kw_only=True, slots=True, frozen=True)
 class BaseConfig:
     service_name: str = "micro-service"
+    service_description: str | None = None
     service_version: str = "1.0.0"
     service_environment: str | None = None
     service_debug: bool = True

lite_bootstrap/fastapi_offline_docs/static/swagger-ui-bundle.js

@@ -1,17 +1,9 @@
 import dataclasses
-import re
-import typing
 
+from lite_bootstrap.helpers import is_valid_path
 from lite_bootstrap.instruments.base import BaseConfig, BaseInstrument
 
 
-VALID_PATH_PATTERN: typing.Final = re.compile(r"^(/[a-zA-Z0-9_-]+)+/?$")
-
-
-def _is_valid_path(maybe_path: str) -> bool:
-    return bool(re.fullmatch(VALID_PATH_PATTERN, maybe_path))
-
-
 @dataclasses.dataclass(kw_only=True, frozen=True)
 class PrometheusConfig(BaseConfig):
     prometheus_metrics_path: str = "/metrics"
@@ -24,6 +16,6 @@ class PrometheusInstrument(BaseInstrument):
     not_ready_message = "prometheus_metrics_path is empty or not valid"
 
     def is_ready(self) -> bool:
-        return bool(self.bootstrap_config.prometheus_metrics_path) and _is_valid_path(
+        return bool(self.bootstrap_config.prometheus_metrics_path) and is_valid_path(
             self.bootstrap_config.prometheus_metrics_path
         )