Add typings

lambdas/handlers/edge_presign_handler.py

@@ -16,14 +16,14 @@
 @handle_edge_exceptions
 @validate_s3_request
 def lambda_handler(event, context):
-    request = event["Records"][0]["cf"]["request"]
+    request: dict = event["Records"][0]["cf"]["request"]
     logger.info("Edge received S3 request")
 
     edge_presign_service = EdgePresignService()
-    request_values = edge_presign_service.extract_request_values(request)
+    request_values: dict = edge_presign_service.extract_request_values(request)
     edge_presign_service.presign_request(request_values)
 
-    request = edge_presign_service.prepare_s3_response(request, request_values)
+    request: dict = edge_presign_service.prepare_s3_response(request, request_values)
 
     logger.info("Edge returning S3 response")
     return request

lambdas/utils/decorators/handle_edge_exceptions.py

@@ -12,7 +12,7 @@
 
 def handle_edge_exceptions(lambda_func: Callable):
     def interceptor(event, context):
-        interaction_id = getattr(request_context, "request_id", None)
+        interaction_id: str | None = getattr(request_context, "request_id", None)
         try:
             return lambda_func(event, context)
         except CloudFrontEdgeException as e:
@@ -24,8 +24,8 @@ def interceptor(event, context):
             ).create_edge_response()
         except Exception as e:
             logger.error(f"Unhandled exception: {str(e)}")
-            err_code = LambdaError.EdgeMalformed.value("err_code")
-            message = LambdaError.EdgeMalformed.value("message")
+            err_code: str = LambdaError.EdgeMalformed.value("err_code")
+            message: str = LambdaError.EdgeMalformed.value("message")
             return EdgeResponse(
                 status_code=500,
                 body=ErrorResponse(err_code, message, interaction_id).create(),

lambdas/utils/decorators/validate_s3_request.py

@@ -24,12 +24,12 @@ def validate_s3_request(lambda_func):
     @wraps(lambda_func)
     def wrapper(event, context):
         request: dict = event["Records"][0]["cf"]["request"]
-
-        if (
+        bad_request: bool = (
             "uri" not in request
             or "querystring" not in request
             or "headers" not in request
-        ):
+        )
+        if bad_request:
             logger.error(
                 "Missing required request components: uri, querystring, or headers."
             )
@@ -45,17 +45,17 @@ def wrapper(event, context):
             logger.error(f"Missing query string: {querystring}")
             raise CloudFrontEdgeException(500, LambdaError.EdgeNoQuery)
 
-        query_params = {
+        query_params: dict[str, str] = {
             query: value[0] for query, value in parse_qs(querystring).items()
         }
-        missing_query_params = [
+        missing_query_params: list[str] = [
             param for param in REQUIRED_QUERY_PARAMS if param not in query_params
         ]
         if missing_query_params:
             logger.error(f"Missing required query parameters: {missing_query_params}")
             raise CloudFrontEdgeException(500, LambdaError.EdgeMissingQuery)
 
-        headers = request["headers"]
+        headers: dict[str, str] = request["headers"]
         missing_headers = [
             header for header in REQUIRED_HEADERS if header.lower() not in headers
         ]