Tidy up FastAPI route functions (#112)

- return objects directly instead of wrapping in response classes
  FastAPI will automatically encode objects as JSON when they are returned
  directly.
  This also means that we can remove orjson -- one less dependency to manage
- use multiple route decorators for monitor functions with same responses
- remove extraneous route function parameters

Other small fixes:
- Ignore bugzilla and atlassian types through pyproject.toml
- Add ServiceHealth type
- Slightly refactor "heartbeat" logic

Author: grahamalama

poetry.lock

@@ -18,7 +18,6 @@ dockerflow = "2022.1.0"
 Jinja2 = "^3.0.3"
 pydantic-yaml = {extras = ["pyyaml","ruamel"], version = "^0.6.1"}
 sentry-sdk = "^1.5.7"
-orjson = "^3.7.7"
 
 
 [tool.poetry.dev-dependencies]
@@ -74,7 +73,7 @@ python_version = "3.10"
 warn_return_any = true
 
 [[tool.mypy.overrides]]
-module = ["ruamel"]
+module = ["ruamel", "bugzilla", "atlassian"]
 ignore_missing_imports = true
 
 [tool.coverage]

pyproject.toml

@@ -11,7 +11,7 @@
 import uvicorn  # type: ignore
 from fastapi import Body, Depends, FastAPI, Request
 from fastapi.encoders import jsonable_encoder
-from fastapi.responses import HTMLResponse, ORJSONResponse
+from fastapi.responses import HTMLResponse
 from fastapi.staticfiles import StaticFiles
 from fastapi.templating import Jinja2Templates
 from sentry_sdk.integrations.asgi import SentryAsgiMiddleware
@@ -49,7 +49,7 @@
 
 
 @app.get("/", include_in_schema=False)
-def root(request: Request):
+def root():
     """Expose key configuration"""
     return {
         "title": app.title,
@@ -98,9 +98,9 @@ def bugzilla_webhook(
     """API endpoint that Bugzilla Webhook Events request"""
     try:
         result = execute_action(request, actions, settings)
-        return ORJSONResponse(content=result, status_code=200)
+        return result
     except IgnoreInvalidRequestError as exception:
-        return ORJSONResponse(content={"error": str(exception)}, status_code=200)
+        return {"error": str(exception)}
 
 
 @app.get("/whiteboard_tags/")

src/app/api.py

@@ -1,61 +1,29 @@
 """
 Router dedicated to Dockerflow APIs
 """
-from typing import Dict
-
-from fastapi import APIRouter, Depends, Request
-from fastapi.responses import ORJSONResponse
+from fastapi import APIRouter, Response
 
 from src.app import environment
 from src.jbi.services import jbi_service_health_map
 
 api_router = APIRouter(tags=["Monitor"])
 
 
-def heartbeat(request: Request, settings: environment.Settings):
-    """Return status of backing services, as required by Dockerflow."""
-    data: Dict = {**jbi_service_health_map()}
-    status_code = 200
-    for _, health in data.items():
-        if not health.get("up"):
-            status_code = 503
-
-    return ORJSONResponse(content=data, status_code=status_code)
-
-
 @api_router.get("/__heartbeat__")
-def get_heartbeat(
-    request: Request,
-    settings: environment.Settings = Depends(environment.get_settings),
-):
-    """Dockerflow API for heartbeat: GET"""
-    return heartbeat(request, settings)
-
-
 @api_router.head("/__heartbeat__")
-def head_heartbeat(
-    request: Request,
-    settings: environment.Settings = Depends(environment.get_settings),
-):
-    """Dockerflow API for heartbeat: HEAD"""
-    return heartbeat(request, settings)
-
-
-def lbheartbeat(request: Request):
-    """Return response when application is running, as required by Dockerflow."""
-    return {"status": "OK"}
+def heartbeat(response: Response):
+    """Return status of backing services, as required by Dockerflow."""
+    health_map = jbi_service_health_map()
+    if not all(health["up"] for health in health_map.values()):
+        response.status_code = 503
+    return health_map
 
 
 @api_router.get("/__lbheartbeat__")
-def get_lbheartbeat(request: Request):
-    """Dockerflow API for lbheartbeat: GET"""
-    return lbheartbeat(request)
-
-
 @api_router.head("/__lbheartbeat__")
-def head_lbheartbeat(request: Request):
+def lbheartbeat():
     """Dockerflow API for lbheartbeat: HEAD"""
-    return lbheartbeat(request)
+    return {"status": "OK"}
 
 
 @api_router.get("/__version__")

src/app/monitor.py

@@ -1,12 +1,17 @@
 """Services and functions that can be used to create custom actions"""
-import bugzilla as rh_bugzilla  # type: ignore
-from atlassian import Jira  # type: ignore
+from typing import TypedDict
+
+import bugzilla as rh_bugzilla
+from atlassian import Jira
 
 from src.app import environment
 
 settings = environment.get_settings()
 
 
+ServiceHealth = TypedDict("ServiceHealth", {"up": bool})
+
+
 def get_jira():
     """Get atlassian Jira Service"""
     return Jira(
@@ -24,18 +29,18 @@ def get_bugzilla():
     )
 
 
-def bugzilla_check_health():
+def bugzilla_check_health() -> ServiceHealth:
     """Check health for Bugzilla Service"""
     bugzilla = get_bugzilla()
-    health = {"up": bugzilla.logged_in}
+    health: ServiceHealth = {"up": bugzilla.logged_in}
     return health
 
 
-def jira_check_health():
+def jira_check_health() -> ServiceHealth:
     """Check health for Jira Service"""
     jira = get_jira()
     server_info = jira.get_server_info(True)
-    health = {"up": server_info is not None}
+    health: ServiceHealth = {"up": server_info is not None}
     return health