refactored the common request interface and added factory method for the mapping

Author: sufyankhanrao

apimatic_core_interfaces/http/request.py

@@ -1,85 +1,211 @@
-from typing import Mapping, Optional, Union, Iterable, Any, Dict
+# request_adapter.py  (Python 3.7+)
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
 
-Headers = Mapping[str, str]
-Cookies = Mapping[str, str]
-QueryParams = Mapping[str, Any]
-FormData = Mapping[str, Any]
-Files = Mapping[str, Any]  # framework-agnostic placeholder (e.g., Starlette UploadFile)
 
+@dataclass(frozen=True)
+class RequestFile:
+    """Framework-agnostic snapshot of an uploaded file (full content)."""
+    filename: str
+    content_type: Optional[str]
+    content: bytes  # full file bytes
 
+
+@dataclass(frozen=True)
 class Request:
     """
-    Framework-agnostic HTTP request snapshot used by verifiers and the webhook manager.
-
-    This class captures all important parts of an HTTP request in a way that does not depend
-    on any specific web framework. It is particularly useful for signature verification and
-    webhook processing.
-
-    Notes:
-        - `body` contains the decoded textual representation of the body (e.g., a JSON string).
-          This should be kept exactly as received.
-        - `raw_body` contains the raw byte stream if available. Use this for cryptographic
-          verification when signatures depend on raw payloads.
-        - Header names should be treated case-insensitively when accessed.
+    Compact, framework-agnostic HTTP request snapshot.
+
+    Fields:
+      - method, path, url
+      - headers: Dict[str, str]      (copied; later mutations on the original won't leak)
+      - raw_body: bytes              (wire bytes; frameworks cache safely)
+      - query/form: Dict[str, List[str]>  (multi-value safe)
+      - cookies: Dict[str, str]
+      - files: Dict[str, List[RequestFile]> (streams read fully & rewound)
     """
+    method: str
+    path: str
+    url: Optional[str]
+    headers: Dict[str, str]
+    raw_body: bytes
+    query: Dict[str, List[str]] = field(default_factory=dict)
+    cookies: Dict[str, str] = field(default_factory=dict)
+    form: Dict[str, List[str]] = field(default_factory=dict)
+    files: Dict[str, List[RequestFile]] = field(default_factory=dict)
+
+    # ---------- helpers ----------
+
+    @staticmethod
+    def _as_listdict(obj: Any) -> Dict[str, List[str]]:
+        """MultiDict/QueryDict → Dict[str, List[str]]; Mapping[str,str] → {k:[v]}."""
+        if not obj:
+            return {}
+        getlist = getattr(obj, "getlist", None)
+        if callable(getlist):
+            return {k: list(getlist(k)) for k in obj.keys()}
+        return {k: [obj[k]] for k in obj.keys()}
+
+    # ---------- factories (non-destructive) ----------
+
+    @staticmethod
+    async def from_fastapi_request(req, *, max_file_bytes: Optional[int] = None) -> "Request":
+        """
+        Build from fastapi.Request / starlette.requests.Request.
+
+        - raw body via await req.body() (Starlette caches; non-destructive)
+        - parse form only for form/multipart content types
+        - read UploadFile content **fully** (or up to max_file_bytes if provided) and rewind stream
+        - copy mappings so later mutations on the original request won't leak
+        """
+        headers = dict(req.headers)
+        raw = await req.body()
+        query = Request._as_listdict(req.query_params)
+        cookies = dict(req.cookies)
+        url_str = str(req.url)
+        path = req.url.path
+
+        ct = (headers.get("content-type") or headers.get("Content-Type") or "").lower()
+        parse_form = ct.startswith(("multipart/form-data", "application/x-www-form-urlencoded"))
+
+        form: Dict[str, List[str]] = {}
+        files: Dict[str, List[RequestFile]] = {}
 
-    # Required fields
-    headers: Optional[Headers] = None
-    method: Optional[str] = None
-    path: Optional[str] = None
-    body: Optional[str] = None
-
-    # Common metadata
-    url: Optional[str] = None
-    query: Optional[QueryParams] = None
-    cookies: Optional[Cookies] = None
-
-    # Optional request bodies
-    raw_body: Optional[bytes] = None
-    form: Optional[FormData] = None
-    files: Optional[Files] = None
-
-    # Arbitrary extensions
-    extensions: Optional[Dict[str, Any]] = None
-
-    def __init__(
-        self,
-        method: Optional[str] = None,
-        path: Optional[str] = None,
-        headers: Optional[Headers] = None,
-        body: Optional[str] = None,
-        url: Optional[str] = None,
-        query: Optional[QueryParams] = None,
-        cookies: Optional[Cookies] = None,
-        raw_body: Optional[bytes] = None,
-        form: Optional[FormData] = None,
-        files: Optional[Files] = None,
-        extensions: Optional[Dict[str, Any]] = None,
-    ):
+        if parse_form:
+            formdata = await req.form()
+            # text fields
+            for k in formdata.keys():
+                for v in formdata.getlist(k):
+                    if not (hasattr(v, "filename") and hasattr(v, "read")):
+                        form.setdefault(k, []).append(str(v))
+            # files (rewind after read)
+            for k in formdata.keys():
+                for v in formdata.getlist(k):
+                    if hasattr(v, "filename") and hasattr(v, "read"):
+                        fobj = getattr(v, "file", None)
+                        # remember current cursor, read all, then rewind
+                        try:
+                            pos = fobj.tell() if fobj else 0
+                        except Exception:
+                            pos = 0
+                        data = await v.read()  # FULL READ
+                        if max_file_bytes is not None and len(data) > max_file_bytes:
+                            data = data[:max_file_bytes]
+                        try:
+                            if fobj:
+                                fobj.seek(pos)
+                        except Exception:
+                            pass
+                        files.setdefault(k, []).append(
+                            RequestFile(v.filename or "", getattr(v, "content_type", None), data)
+                        )
+
+        return Request(
+            method=req.method,
+            path=path,
+            url=url_str,
+            headers=headers,
+            raw_body=raw,
+            query=query,
+            cookies=cookies,
+            form=form,
+            files=files,
+        )
+
+    @staticmethod
+    def from_django_request(req, *, max_file_bytes: Optional[int] = None) -> "Request":
         """
-        Initialize a new request snapshot.
-
-        Args:
-            method: HTTP method of the request (e.g., "GET", "POST").
-            path: URL path of the request, excluding query parameters.
-            headers: Mapping of header keys to values (case-insensitive).
-            body: Decoded body content as a string, such as a JSON or XML string.
-            url: Full request URL if available.
-            query: Mapping of query parameter keys to single or multiple values.
-            cookies: Mapping of cookie keys to values.
-            raw_body: Raw bytes of the request body, for signature verification.
-            form: Parsed form data, if the request contains form fields.
-            files: Uploaded files associated with the request.
-            extensions: Arbitrary framework-specific or user-defined metadata.
+        Build from django.http.HttpRequest.
+
+        - uses req.body (cached bytes; non-destructive)
+        - text fields from req.POST; files from req.FILES (read FULL & rewind)
+        - copies mappings to avoid leaking later mutations
         """
-        self.method = method
-        self.path = path
-        self.headers = headers
-        self.body = body
-        self.url = url
-        self.query = query
-        self.cookies = cookies
-        self.raw_body = raw_body
-        self.form = form
-        self.files = files
-        self.extensions = extensions
+        headers = dict(getattr(req, "headers", {}) or {})
+        url_str = req.build_absolute_uri()
+        path = req.path
+        raw = bytes(getattr(req, "body", b"") or b"")
+        query = Request._as_listdict(getattr(req, "GET", {}))
+        cookies = dict(getattr(req, "COOKIES", {}) or {})
+
+        form = Request._as_listdict(getattr(req, "POST", {}))
+        files: Dict[str, List[RequestFile]] = {}
+        files_src = getattr(req, "FILES", None)
+        if files_src and hasattr(files_src, "getlist"):
+            for k in files_src.keys():
+                for f in files_src.getlist(k):
+                    try:
+                        pos = f.tell()
+                    except Exception:
+                        pos = 0
+                    data = f.read()  # FULL READ
+                    if max_file_bytes is not None and len(data) > max_file_bytes:
+                        data = data[:max_file_bytes]
+                    try:
+                        f.seek(pos)
+                    except Exception:
+                        pass
+                    files.setdefault(k, []).append(
+                        RequestFile(getattr(f, "name", "") or "", getattr(f, "content_type", None), data)
+                    )
+
+        return Request(
+            method=req.method,
+            path=path,
+            url=url_str,
+            headers=headers,
+            raw_body=raw,
+            query=query,
+            cookies=cookies,
+            form=form,
+            files=files,
+        )
+
+    @staticmethod
+    def from_flask_request(req, *, max_file_bytes: Optional[int] = None) -> "Request":
+        """
+        Build from flask.Request (Werkzeug).
+
+        - uses req.get_data(cache=True) (non-destructive)
+        - text fields from req.form; files from req.files (read FULL & rewind)
+        - copies mappings to avoid leaking later mutations
+        """
+        headers = dict(req.headers)
+        url_str = getattr(req, "url", None)
+        path = req.path
+        raw = req.get_data(cache=True)
+        query = Request._as_listdict(req.args)
+        cookies = dict(req.cookies)
+
+        form = Request._as_listdict(req.form)
+        files: Dict[str, List[RequestFile]] = {}
+        for k in req.files.keys():
+            for s in req.files.getlist(k):
+                stream = getattr(s, "stream", None)
+                try:
+                    pos = stream.tell() if stream else 0
+                except Exception:
+                    pos = 0
+                data = s.read()  # FULL READ
+                if max_file_bytes is not None and len(data) > max_file_bytes:
+                    data = data[:max_file_bytes]
+                try:
+                    if stream:
+                        stream.seek(pos)
+                except Exception:
+                    pass
+                files.setdefault(k, []).append(
+                    RequestFile(s.filename or "", getattr(s, "mimetype", None), data)
+                )
+
+        return Request(
+            method=req.method,
+            path=path,
+            url=url_str,
+            headers=headers,
+            raw_body=raw,
+            query=query,
+            cookies=cookies,
+            form=form,
+            files=files,
+        )