feat: add function to build aws api-gateway aws-proxy event structure (…

…#29)

src/routing.py

@@ -1,7 +1,9 @@
 import json
 import logging
+from datetime import datetime, timezone
 from http import HTTPStatus
 from typing import Any, Callable, Dict
+from uuid import uuid4
 
 from fastapi import Request, Response, routing
 
@@ -10,6 +12,35 @@
 Handler = Callable[[Dict[str, Any], Any], Dict[str, Any]]
 
 
+async def event_builder(request: Request) -> Dict[str, Any]:
+    now = datetime.now(timezone.utc)
+    body = await request.body()
+    event = {
+        "body": body.decode(),
+        "path": request.url.path,
+        "httpMethod": request.method,
+        "isBase64Encoded": False,
+        "queryStringParameters": dict(request.query_params),
+        "pathParameters": dict(request.path_params),
+        "headers": dict(request.headers),
+        "requestContext": {
+            "stage": request.app.version,
+            "requestId": str(uuid4()),
+            "requestTime": now.strftime(r"%d/%b/%Y:%H:%M:%S %z"),
+            "requestTimeEpoch": int(now.timestamp()),
+            "identity": {
+                "sourceIp": getattr(request.client, "host", None),
+                "userAgent": request.headers.get("user-agent"),
+            },
+            "path": request.url.path,
+            "httpMethod": request.method,
+            "protocol": f"HTTP/{request.scope['http_version']}",
+        },
+    }
+
+    return event
+
+
 def default_endpoint(request: Request) -> Response:
     logger.error(f"Executing default endpoint: {request.scope}")
 

tests/test_routing.py

@@ -1,8 +1,9 @@
-from http import HTTPStatus
 import unittest
+from http import HTTPStatus
+
+from fastapi import FastAPI, Request, Response
 
 import routing
-from fastapi import Response, Request
 
 
 class TestAPIRoute(unittest.TestCase):
@@ -45,3 +46,41 @@ def test_import_handler(self):
         self.assertTrue(callable(handler))
         self.assertEqual(getattr(handler, "__module__", None), module_name)
         self.assertEqual(getattr(handler, "__name__", None), handler_name)
+
+
+class TestEventBuilder(unittest.IsolatedAsyncioTestCase):
+    async def test_event_builder(self):
+        async def receive():
+            return {"type": "http.request", "body": b'{"message":  "test"}'}
+
+        scope = {
+            "type": "http",
+            "http_version": "1.1",
+            "root_path": "",
+            "path": "/test",
+            "method": "GET",
+            "query_string": [],
+            "path_params": {},
+            "client": ("127.0.0.1", 80),
+            "app": FastAPI(),
+            "headers": [
+                (b"content-type", b"application/json"),
+                (b"user-agent", b"python/unittest"),
+            ],
+        }
+        request = Request(scope, receive)
+        expected_keys = {
+            "body",
+            "path",
+            "httpMethod",
+            "isBase64Encoded",
+            "queryStringParameters",
+            "pathParameters",
+            "headers",
+            "requestContext",
+        }
+
+        event = await routing.event_builder(request)
+
+        self.assertIsInstance(event, dict)
+        self.assertEqual(set(event.keys()), expected_keys)