Add core function structure based on fastapi

Author: marvinbuss

code/function/.gitkeep renamed to code/function/api/__init__.py

@@ -0,0 +1,8 @@
+from fastapi import APIRouter
+from function.api.v1.endpoints import heartbeat, sample
+
+api_v1_router = APIRouter()
+api_v1_router.include_router(
+    sample.router, prefix="/landingZone", tags=["sample"]
+)
+api_v1_router.include_router(heartbeat.router, prefix="/health", tags=["health"])

code/function/api/v1/__init__.py

@@ -0,0 +1,15 @@
+from typing import Any
+
+from fastapi import APIRouter
+from function.models.heartbeat import HearbeatResult
+from function.utils import setup_logging
+
+logger = setup_logging(__name__)
+
+router = APIRouter()
+
+
+@router.get("/heartbeat", response_model=HearbeatResult, name="heartbeat")
+async def get_hearbeat() -> Any:
+    logger.info("Received Heartbeat Request")
+    return HearbeatResult(isAlive=True)

code/function/api/v1/api_v1.py

@@ -0,0 +1,20 @@
+from typing import Any
+
+from fastapi import APIRouter
+from function.models.sample import (
+    SampleRequest,
+    SampleResponse,
+)
+from function.utils import setup_logging
+
+logger = setup_logging(__name__)
+
+router = APIRouter()
+
+
+@router.post("/create", response_model=SampleResponse, name="create")
+async def post_predict(
+    data: SampleRequest,
+) -> SampleResponse:
+    logger.info(f"Received request: {data}")
+    return SampleResponse(output=f"Hello ${data.input}")

code/function/api/v1/endpoints/heartbeat.py

@@ -0,0 +1,15 @@
+import logging
+
+from pydantic import BaseSettings
+
+
+class Settings(BaseSettings):
+    PROJECT_NAME: str = "FunctionSample"
+    SERVER_NAME: str = "FunctionSample"
+    APP_VERSION: str = "v0.0.1"
+    API_V1_STR: str = "/v1"
+    LOGGING_LEVEL: int = logging.INFO
+    DEBUG: bool = False
+
+
+settings = Settings()

code/function/api/v1/endpoints/sample.py

@@ -0,0 +1,14 @@
+from pydantic import BaseSettings
+
+
+class Messages(BaseSettings):
+    # Base messages
+    NO_API_KEY = "No API key provided."
+    AUTH_REQ = "Authentication required."
+    HTTP_500_DETAIL = "Internal server error."
+
+    # Templates
+    NO_VALID_PAYLOAD = "{} is not a valid payload."
+
+
+messages = Messages()

code/function/core/__init__.py

@@ -1,29 +1,34 @@
-import logging
-
 import azure.functions as func
+from fastapi import FastAPI
+from function.api.v1.api_v1 import api_v1_router
+from function.core.config import settings
+
+
+def get_app() -> FastAPI:
+    app = FastAPI(
+        title=settings.PROJECT_NAME,
+        version=settings.APP_VERSION,
+        openapi_url="/openapi.json",
+        debug=settings.DEBUG,
+    )
+    app.include_router(api_v1_router, prefix=settings.API_V1_STR)
+    return app
+
+
+fastapi_app = get_app()
+
+
+@fastapi_app.on_event("startup")
+async def startup_event():
+    pass
+
+
+@fastapi_app.on_event("shutdown")
+async def shutdown_event():
+    pass
+
 
-app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)
-
-
-@app.route(route="HttpTrigger")
-def HttpTrigger(req: func.HttpRequest) -> func.HttpResponse:
-    logging.info("Python HTTP trigger function processed a request.")
-
-    name = req.params.get("name")
-    if not name:
-        try:
-            req_body = req.get_json()
-        except ValueError:
-            pass
-        else:
-            name = req_body.get("name")
-
-    if name:
-        return func.HttpResponse(
-            f"Hello, {name}. This HTTP triggered function executed successfully."
-        )
-    else:
-        return func.HttpResponse(
-            "This HTTP triggered function executed successfully. Pass a name in the query string or in the request body for a personalized response.",
-            status_code=200,
-        )
+app = func.AsgiFunctionApp(
+    app=fastapi_app,
+    http_auth_level=func.AuthLevel.FUNCTION,
+)