initial migration to litestar

hyperglass/api/__init__.py

@@ -1,253 +1,62 @@
-"""hyperglass REST API & Web UI."""
-
-# Standard Library
-import sys
-from typing import List
-from pathlib import Path
+"""hyperglass API."""
 
 # Third Party
-from fastapi import FastAPI
-from fastapi.responses import JSONResponse
-from fastapi.exceptions import ValidationException, RequestValidationError
-from fastapi.staticfiles import StaticFiles
-from starlette.exceptions import HTTPException as StarletteHTTPException
-from fastapi.openapi.utils import get_openapi
-from fastapi.middleware.cors import CORSMiddleware
-from fastapi.middleware.gzip import GZipMiddleware
+from litestar import Litestar
+from litestar.openapi import OpenAPIConfig
+from litestar.exceptions import HTTPException, ValidationException
+from litestar.static_files import create_static_files_router
 
 # Project
-from hyperglass.log import log
-from hyperglass.util import cpu_count
 from hyperglass.state import use_state
 from hyperglass.constants import __version__
-from hyperglass.models.ui import UIParameters
-from hyperglass.api.events import on_startup, on_shutdown
-from hyperglass.api.routes import docs, info, query, router, queries, routers, ui_props
 from hyperglass.exceptions import HyperglassError
-from hyperglass.api.error_handlers import (
-    app_handler,
-    http_handler,
-    default_handler,
-    validation_handler,
-)
-from hyperglass.models.api.response import (
-    QueryError,
-    InfoResponse,
-    QueryResponse,
-    RoutersResponse,
-    SupportedQueryResponse,
-)
 
-STATE = use_state()
+# Local
+from .events import check_redis
+from .routes import info, query, device, devices, queries
+from .middleware import COMPRESSION_CONFIG, create_cors_config
+from .error_handlers import app_handler, http_handler, default_handler, validation_handler
+
+__all__ = ("app",)
 
-WORKING_DIR = Path(__file__).parent
-EXAMPLES_DIR = WORKING_DIR / "examples"
+STATE = use_state()
 
 UI_DIR = STATE.settings.static_path / "ui"
 IMAGES_DIR = STATE.settings.static_path / "images"
 
-EXAMPLE_DEVICES_PY = EXAMPLES_DIR / "devices.py"
-EXAMPLE_QUERIES_PY = EXAMPLES_DIR / "queries.py"
-EXAMPLE_QUERY_PY = EXAMPLES_DIR / "query.py"
-EXAMPLE_DEVICES_CURL = EXAMPLES_DIR / "devices.sh"
-EXAMPLE_QUERIES_CURL = EXAMPLES_DIR / "queries.sh"
-EXAMPLE_QUERY_CURL = EXAMPLES_DIR / "query.sh"
 
-ASGI_PARAMS = {
-    "host": str(STATE.settings.host),
-    "port": STATE.settings.port,
-    "debug": STATE.settings.debug,
-    "workers": cpu_count(2),
-}
-DOCS_PARAMS = {}
-if STATE.params.docs.enable:
-    DOCS_PARAMS.update({"openapi_url": "/openapi.json"})
-    if STATE.params.docs.mode == "redoc":
-        DOCS_PARAMS.update({"docs_url": None, "redoc_url": STATE.params.docs.path})
-    elif STATE.params.docs.mode == "swagger":
-        DOCS_PARAMS.update({"docs_url": STATE.params.docs.path, "redoc_url": None})
-
-for directory in (UI_DIR, IMAGES_DIR):
-    if not directory.exists():
-        log.warning("Directory '{d}' does not exist, creating...", d=str(directory))
-        directory.mkdir()
-
-# Main App Definition
-app = FastAPI(
-    debug=STATE.settings.debug,
-    title=STATE.params.site_title,
-    description=STATE.params.site_description,
+OPEN_API = OpenAPIConfig(
+    title=STATE.params.docs.title.format(site_title=STATE.params.site_title),
     version=__version__,
-    default_response_class=JSONResponse,
-    **DOCS_PARAMS,
-)
-
-# Add Event Handlers
-for startup in on_startup:
-    app.add_event_handler("startup", startup)
-
-for shutdown in on_shutdown:
-    app.add_event_handler("shutdown", shutdown)
-
-# HTTP Error Handler
-app.add_exception_handler(StarletteHTTPException, http_handler)
-
-# Backend Application Error Handler
-app.add_exception_handler(HyperglassError, app_handler)
-
-# Request Validation Error Handler
-app.add_exception_handler(RequestValidationError, validation_handler)
-
-# App Validation Error Handler
-app.add_exception_handler(ValidationException, validation_handler)
-
-# Uncaught Error Handler
-app.add_exception_handler(Exception, default_handler)
-
-
-def _custom_openapi():
-    """Generate custom OpenAPI config."""
-    openapi_schema = get_openapi(
-        title=STATE.params.docs.title.format(site_title=STATE.params.site_title),
-        version=__version__,
-        description=STATE.params.docs.description,
-        routes=app.routes,
-    )
-    openapi_schema["info"]["x-logo"] = {"url": "/images/light" + STATE.params.web.logo.light.suffix}
-
-    query_samples = []
-    queries_samples = []
-    devices_samples = []
-
-    with EXAMPLE_QUERY_CURL.open("r") as e:
-        example = e.read()
-        query_samples.append({"lang": "cURL", "source": example % str(STATE.params.docs.base_url)})
-
-    with EXAMPLE_QUERY_PY.open("r") as e:
-        example = e.read()
-        query_samples.append(
-            {"lang": "Python", "source": example % str(STATE.params.docs.base_url)}
-        )
-
-    with EXAMPLE_DEVICES_CURL.open("r") as e:
-        example = e.read()
-        queries_samples.append(
-            {"lang": "cURL", "source": example % str(STATE.params.docs.base_url)}
-        )
-    with EXAMPLE_DEVICES_PY.open("r") as e:
-        example = e.read()
-        queries_samples.append(
-            {"lang": "Python", "source": example % str(STATE.params.docs.base_url)}
-        )
-
-    with EXAMPLE_QUERIES_CURL.open("r") as e:
-        example = e.read()
-        devices_samples.append(
-            {"lang": "cURL", "source": example % str(STATE.params.docs.base_url)}
-        )
-
-    with EXAMPLE_QUERIES_PY.open("r") as e:
-        example = e.read()
-        devices_samples.append(
-            {"lang": "Python", "source": example % str(STATE.params.docs.base_url)}
-        )
-
-    openapi_schema["paths"]["/api/query/"]["post"]["x-code-samples"] = query_samples
-    openapi_schema["paths"]["/api/devices"]["get"]["x-code-samples"] = devices_samples
-    openapi_schema["paths"]["/api/queries"]["get"]["x-code-samples"] = queries_samples
-
-    app.openapi_schema = openapi_schema
-    return app.openapi_schema
-
-
-CORS_ORIGINS = STATE.params.cors_origins.copy()
-if STATE.settings.dev_mode:
-    CORS_ORIGINS = [*CORS_ORIGINS, STATE.settings.dev_url, "http://localhost:3000"]
-
-# CORS Configuration
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=CORS_ORIGINS,
-    allow_methods=["GET", "POST", "OPTIONS"],
-    allow_headers=["*"],
-)
-
-# GZIP Middleware
-app.add_middleware(GZipMiddleware)
-
-app.add_api_route(
-    path="/api/info",
-    endpoint=info,
-    methods=["GET"],
-    response_model=InfoResponse,
-    response_class=JSONResponse,
-    summary=STATE.params.docs.info.summary,
-    description=STATE.params.docs.info.description,
-    tags=[STATE.params.docs.info.title],
-)
-
-app.add_api_route(
-    path="/api/devices",
-    endpoint=routers,
-    methods=["GET"],
-    response_model=List[RoutersResponse],
-    response_class=JSONResponse,
-    summary=STATE.params.docs.devices.summary,
-    description=STATE.params.docs.devices.description,
-    tags=[STATE.params.docs.devices.title],
-)
-
-app.add_api_route(
-    path="/api/devices/{id}",
-    endpoint=router,
-    methods=["GET"],
-    response_model=RoutersResponse,
-    response_class=JSONResponse,
-    summary=STATE.params.docs.devices.summary,
-    description=STATE.params.docs.devices.description,
-    tags=[STATE.params.docs.devices.title],
-)
-
-app.add_api_route(
-    path="/api/queries",
-    endpoint=queries,
-    methods=["GET"],
-    response_class=JSONResponse,
-    response_model=List[SupportedQueryResponse],
-    summary=STATE.params.docs.queries.summary,
-    description=STATE.params.docs.queries.description,
-    tags=[STATE.params.docs.queries.title],
-)
-
-app.add_api_route(
-    path="/api/query",
-    endpoint=query,
-    methods=["POST"],
-    summary=STATE.params.docs.query.summary,
-    description=STATE.params.docs.query.description,
-    responses={
-        400: {"model": QueryError, "description": "Request Content Error"},
-        422: {"model": QueryError, "description": "Request Format Error"},
-        500: {"model": QueryError, "description": "Server Error"},
+    description=STATE.params.docs.description,
+    path=STATE.params.docs.path,
+    root_schema_site="elements",
+)
+
+
+app = Litestar(
+    route_handlers=[
+        device,
+        devices,
+        queries,
+        info,
+        query,
+        create_static_files_router(
+            path="/images", directories=[IMAGES_DIR], name="images", include_in_schema=False
+        ),
+        create_static_files_router(
+            path="/", directories=[UI_DIR], name="ui", html_mode=True, include_in_schema=False
+        ),
+    ],
+    exception_handlers={
+        HTTPException: http_handler,
+        HyperglassError: app_handler,
+        ValidationException: validation_handler,
+        Exception: default_handler,
     },
-    response_model=QueryResponse,
-    tags=[STATE.params.docs.query.title],
-    response_class=JSONResponse,
-)
-
-app.add_api_route(
-    path="/ui/props/",
-    endpoint=ui_props,
-    methods=["GET", "OPTIONS"],
-    response_class=JSONResponse,
-    response_model=UIParameters,
-    response_model_by_alias=True,
+    on_startup=[check_redis],
+    debug=STATE.settings.debug,
+    cors_config=create_cors_config(state=STATE),
+    compression_config=COMPRESSION_CONFIG,
+    openapi_config=OPEN_API if STATE.params.docs.enable else None,
 )
-
-
-if STATE.params.docs.enable:
-    app.add_api_route(path=STATE.params.docs.path, endpoint=docs, include_in_schema=False)
-    app.openapi = _custom_openapi
-
-app.mount("/images", StaticFiles(directory=IMAGES_DIR), name="images")
-app.mount("/", StaticFiles(directory=UI_DIR, html=True), name="ui")

hyperglass/api/error_handlers.py

@@ -1,49 +1,55 @@
 """API Error Handlers."""
 
 # Third Party
-from fastapi import Request
-from starlette.responses import JSONResponse
+from litestar import Request, Response
 
 # Project
 from hyperglass.log import log
 from hyperglass.state import use_state
 
+__all__ = (
+    "default_handler",
+    "http_handler",
+    "app_handler",
+    "validation_handler",
+)
 
-async def default_handler(request: Request, exc: BaseException) -> JSONResponse:
+
+def default_handler(request: Request, exc: BaseException) -> Response:
     """Handle uncaught errors."""
     state = use_state()
     log.critical(
         "{method} {path} {detail!s}", method=request.method, path=request.url.path, detail=exc
     )
-    return JSONResponse(
+    return Response(
         {"output": state.params.messages.general, "level": "danger", "keywords": []},
         status_code=500,
     )
 
 
-async def http_handler(request: Request, exc: BaseException) -> JSONResponse:
+def http_handler(request: Request, exc: BaseException) -> Response:
     """Handle web server errors."""
     log.critical(
         "{method} {path} {detail}", method=request.method, path=request.url.path, detail=exc.detail
     )
-    return JSONResponse(
+    return Response(
         {"output": exc.detail, "level": "danger", "keywords": []},
         status_code=exc.status_code,
     )
 
 
-async def app_handler(request: Request, exc: BaseException) -> JSONResponse:
+def app_handler(request: Request, exc: BaseException) -> Response:
     """Handle application errors."""
     log.critical(
         "{method} {path} {detail}", method=request.method, path=request.url.path, detail=exc.message
     )
-    return JSONResponse(
+    return Response(
         {"output": exc.message, "level": exc.level, "keywords": exc.keywords},
         status_code=exc.status_code,
     )
 
 
-async def validation_handler(request: Request, exc: BaseException) -> JSONResponse:
+def validation_handler(request: Request, exc: BaseException) -> Response:
     """Handle Pydantic validation errors raised by FastAPI."""
     error = exc.errors()[0]
     log.critical(
@@ -52,7 +58,7 @@ async def validation_handler(request: Request, exc: BaseException) -> JSONRespon
         path=request.url.path,
         detail=error["msg"],
     )
-    return JSONResponse(
+    return Response(
         {"output": error["msg"], "level": "error", "keywords": error["loc"]},
         status_code=422,
     )

hyperglass/api/events.py

@@ -1,14 +1,18 @@
 """API Events."""
 
+# Standard Library
+import typing as t
+
+# Third Party
+from litestar import Litestar
+
 # Project
 from hyperglass.state import use_state
 
+__all__ = ("check_redis",)
+
 
-def check_redis() -> None:
+async def check_redis(_: Litestar) -> t.NoReturn:
     """Ensure Redis is running before starting server."""
     cache = use_state("cache")
     cache.check()
-
-
-on_startup = (check_redis,)
-on_shutdown = ()