Merge pull request #2 from surquest/fix

removing hard dependency on GCP

Author: surquest

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "surquest-fastapi-utils"
-version = "0.2.1"
+version = "0.2.3"
 description = "This project provides collection of utilities for FastAPI framework as: Catcher, Middleware, etc."
 authors = [
     {name= "Michal Švarc", email= "michal.svarc@surquest.com"}

src/surquest/fastapi/utils/GCP/catcher.py

@@ -124,6 +124,6 @@ async def catch_http_exception(
         )
 
         return Response.set(
-            status_code=404,
+            status_code=getattr(exc, "status_code", 404),
             errors=[message]
         )

src/surquest/fastapi/utils/GCP/tracer.py

@@ -18,13 +18,21 @@
 
 tracer_provider = TracerProvider(sampler=ALWAYS_ON)  # always trace
 trace.set_tracer_provider(tracer_provider)
-cloud_trace_exporter = CloudTraceSpanExporter()
-tracer_provider.add_span_processor(
-    # BatchSpanProcessor buffers spans and sends them in batches in a
-    # background thread. The default parameters are sensible, but can be
-    # tweaked to optimize your performance
-    BatchSpanProcessor(cloud_trace_exporter)
-)
+# cloud_trace_exporter = CloudTraceSpanExporter()
+try:
+    cloud_trace_exporter = CloudTraceSpanExporter()
+except Exception as e:
+    print("CloudTraceSpanExporter not initialized")
+    print(e)
+    cloud_trace_exporter = None
+
+if cloud_trace_exporter is not None:    
+    tracer_provider.add_span_processor(
+        # BatchSpanProcessor buffers spans and sends them in batches in a
+        # background thread. The default parameters are sensible, but can be
+        # tweaked to optimize your performance
+        BatchSpanProcessor(cloud_trace_exporter)
+    )
 
 class Tracer:
 

src/surquest/fastapi/utils/auth.py

@@ -0,0 +1,38 @@
+import typing
+from fastapi.security.api_key import APIKeyQuery
+from fastapi import HTTPException, Security, Request
+
+
+api_key_query = APIKeyQuery(name="apiKey", auto_error=False)
+
+async def check_api_key(
+    request: Request,
+    api_key_query: str = Security(api_key_query),
+):
+
+    # check if app has defined API key
+    if "api_key" in request.app.extra:
+
+        # check if passed API key is equal to app's API key
+        if api_key_query == request.app.extra.get("api_key"):
+            return api_key_query
+        else:
+            raise AUTHException()
+
+
+class AUTHException(HTTPException):
+    def __init__(
+        self,
+        status_code: int = 401, # HTTP status code for Unauthorized
+        detail: typing.Optional[str] = "Unauthorized: Invalid API Key",
+        loc: typing.Optional[list] = ["query", "apiKey"],
+        type: typing.Optional[str] = "AUTH.ERROR"
+    ) -> None:
+        self.status_code = status_code
+        self.detail = detail
+        self.loc = loc
+        self.type = type
+
+    def __repr__(self) -> str:
+        class_name = self.__class__.__name__
+        return f"{class_name}(status_code={self.status_code!r}, detail={self.detail!r})"