feat: generate api gateway proxy event from http request (#209)

* Create a module with helper functions to generate AWS events for Lambda functions.
  * Add a function that generates an event of type ApiGateway proxy.

faster_sam/dependencies/events.py

@@ -0,0 +1,33 @@
+from datetime import datetime, timezone
+from typing import Any, Dict
+from uuid import uuid4
+
+from fastapi import Request
+
+
+async def apigateway_proxy(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

tests/test_dependencies_events.py

@@ -0,0 +1,52 @@
+import unittest
+
+from fastapi import FastAPI, Request
+
+from faster_sam.dependencies import events
+
+
+def build_request():
+    async def receive():
+        return {"type": "http.request", "body": b'{"message": "pong"}'}
+
+    scope = {
+        "type": "http",
+        "http_version": "1.1",
+        "root_path": "",
+        "path": "/ping/pong",
+        "method": "GET",
+        "query_string": b"q=all&skip=100",
+        "path_params": {"message": "pong"},
+        "client": ("127.0.0.1", 80),
+        "app": FastAPI(),
+        "headers": [
+            (b"content-type", b"application/json"),
+            (b"user-agent", b"python/unittest"),
+        ],
+    }
+
+    return Request(scope, receive)
+
+
+class TestApiGatewayProxy(unittest.IsolatedAsyncioTestCase):
+    async def test_event(self):
+        request = build_request()
+        event = await events.apigateway_proxy(request)
+
+        self.assertIsInstance(event, dict)
+        self.assertEqual(event["body"], '{"message": "pong"}')
+        self.assertEqual(event["path"], "/ping/pong")
+        self.assertEqual(event["httpMethod"], "GET")
+        self.assertEqual(event["isBase64Encoded"], False)
+        self.assertEqual(event["queryStringParameters"], {"q": "all", "skip": "100"})
+        self.assertEqual(event["pathParameters"], {"message": "pong"})
+        self.assertEqual(
+            event["headers"],
+            {"content-type": "application/json", "user-agent": "python/unittest"},
+        )
+        self.assertEqual(event["requestContext"]["stage"], "0.1.0")
+        self.assertEqual(event["requestContext"]["identity"]["sourceIp"], "127.0.0.1")
+        self.assertEqual(event["requestContext"]["identity"]["userAgent"], "python/unittest")
+        self.assertEqual(event["requestContext"]["path"], "/ping/pong")
+        self.assertEqual(event["requestContext"]["httpMethod"], "GET")
+        self.assertEqual(event["requestContext"]["protocol"], "HTTP/1.1")