Migrate to fastApi (#154)

Dockerfile

@@ -13,6 +13,6 @@ RUN pip install --no-cache-dir -r requirements.txt -r semantic_search_requiremen
 
 COPY ./src ./src
 
-CMD exec gunicorn --bind :$PORT --workers 2 --threads 8 --timeout 600 -k gevent index:flask_app --chdir /app/src
+CMD exec uvicorn index:api --host 0.0.0.0 --port $PORT --workers 2 --timeout-keep-alive 600
 
 EXPOSE 8080

requirements.txt

@@ -4,3 +4,5 @@ slack-bolt==1.18.0
 openai==0.28
 gunicorn==20.1.0
 gevent==23.9.1
+fastapi
+uvicorn

src/index.py

@@ -1,34 +1,44 @@
+from fastapi import FastAPI, Request, BackgroundTasks
+from pydantic import BaseModel
 import os
 import logging
-from flask import Flask, request
-from slack_bolt.adapter.flask import SlackRequestHandler
-from lib.guards import shared_secret_guard
-from semantic_search.semantic_search.external_services import openai
+from slack_bolt.adapter.fastapi import SlackRequestHandler
+from slack_bolt import App
+from fastapi.responses import JSONResponse
 from semantic_search.semantic_search.query import smart_query
 from services.slack_service import slack_app, handle_app_installed
 
-logging.basicConfig(level=os.environ["LOG_LEVEL"])
 
+# Configuration and Logging
+LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO")
+logging.basicConfig(level=LOG_LEVEL)
 
-flask_app = Flask("Haly")
-handler = SlackRequestHandler(slack_app)
+# Initialize FastAPI app and Slack Bolt App
+api = FastAPI()
+app_handler = SlackRequestHandler(slack_app)
 
-@flask_app.route("/slack/events", methods=["POST"])
-def slack_events():
-    return handler.handle(request)
+# Routes
+@api.post("/slack/events")
+async def slack_events_endpoint(req: Request):
+    return await app_handler.handle(req)
 
-@flask_app.route("/slack/app-installed", methods=["POST"])
-@shared_secret_guard
-def app_installed_route():
-    return handle_app_installed(request)
+@api.post("/slack/app-installed")
+async def app_installed_route(req: Request):
+    handle_app_installed(req)
+    return JSONResponse(content={"message": "App installed successfully"})
 
-@flask_app.route("/test-smart-query", methods=["POST"])
-def test_smart_query():
-    return {
-        "response": smart_query("T03QUQ2NFQC", request.json["query"], request.json["name"])
-    }
+# Pydantic model for smart query requests
+class SmartQueryRequest(BaseModel):
+    query: str
+    name: str
+@api.post("/test-smart-query")
+async def test_smart_query(req: SmartQueryRequest):
+    # Using Pydantic model for automatic request parsing and validation
+    # ... code for smart query ...
+    return {"response": smart_query("T03QUQ2NFQC", req.query, req.name)}
 
 
+# Running the application with Uvicorn
 if __name__ == "__main__":
-    openai.bootstrap()
-    flask_app.run(debug=True, host="0.0.0.0", port=int(os.environ.get("PORT", 8080)))
+    import uvicorn
+    uvicorn.run(api, host="0.0.0.0", port=int(os.environ.get("PORT", 8000)))

src/lib/guards.py

@@ -1,45 +1,34 @@
 from functools import wraps
-from http import HTTPStatus
+from fastapi import HTTPException, Request, status
+from fastapi.responses import JSONResponse
 import os
 import time
 
-from flask import abort, jsonify, request
-
 unauthorized_error = {
     "message": "Requires authentication"
 }
 
 def time_tracker(func):
-    def wrapper(*args, **kwargs):
+    @wraps(func)
+    async def wrapper(*args, **kwargs):
         start_time = time.perf_counter()
-        result = func(*args, **kwargs)
+        result = await func(*args, **kwargs)  # Assuming this can be an async function
         end_time = time.perf_counter()
         elapsed_time = end_time - start_time
-        print(
-            f"Function '{func.__name__}' took {elapsed_time:.4f} seconds to execute.")
+        print(f"Function '{func.__name__}' took {elapsed_time:.4f} seconds to execute.")
         return result
-
     return wrapper
 
-def shared_secret_guard(function):
-    @wraps(function)
-    def decorator(*args, **kwargs):
-        secret = get_secret_from_request()
-        if secret != os.environ["API_SHARED_SECRET"]:
-            json_abort(HTTPStatus.UNAUTHORIZED, unauthorized_error)
-            return None
-        return function(*args, **kwargs)
+def shared_secret_guard(request: Request):
+    secret = get_secret_from_request(request)
+    if secret != os.environ["API_SHARED_SECRET"]:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=unauthorized_error)
 
-    return decorator
+    # If the secret matches, simply return True or some other success indicator
+    return True
 
-def get_secret_from_request():
-    secret = request.headers.get("X-Shared-Secret", None)
+def get_secret_from_request(request: Request):
+    secret = request.headers.get("X-Shared-Secret")
     if not secret:
-        json_abort(HTTPStatus.UNAUTHORIZED, unauthorized_error)
-        return None
-    return secret
-
-def json_abort(status_code, data=None):
-    response = jsonify(data)
-    response.status_code = status_code
-    abort(response)
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=unauthorized_error)
+    return secret