revision: adds structural typing and adapter for requests

Author: sufyankhanrao

apimatic_core/adapters/__init__.py

@@ -0,0 +1,6 @@
+__all__ = [
+    'request_adapter',
+    'flask_like',
+    'django_like',
+    'starlette_like'
+]

apimatic_core/adapters/django_like.py

@@ -0,0 +1,13 @@
+from typing import Any, Mapping
+from typing_extensions import Protocol, runtime_checkable
+
+@runtime_checkable
+class DjangoRequestLike(Protocol):
+    method: str
+    headers: Mapping[str, str]
+    COOKIES: Mapping[str, str]
+    GET: Mapping[str, Any]
+    POST: Mapping[str, Any]
+    path: str
+    body: bytes
+    def build_absolute_uri(self) -> str: ...

apimatic_core/adapters/flask_like.py

@@ -0,0 +1,13 @@
+from typing import Any, Mapping
+from typing_extensions import Protocol, runtime_checkable
+
+@runtime_checkable
+class FlaskRequestLike(Protocol):
+    method: str
+    headers: Mapping[str, str]
+    cookies: Mapping[str, str]
+    args: Mapping[str, Any]
+    url: str
+    path: str
+    def get_data(self, cache: bool = ...) -> bytes: ...
+    form: Mapping[str, Any]

apimatic_core/adapters/request_adapter.py

@@ -0,0 +1,133 @@
+import asyncio
+from typing import Dict, List, Optional, Union, Any
+from http.cookies import SimpleCookie
+from apimatic_core_interfaces.http.request import Request
+
+from apimatic_core.adapters.django_like import DjangoRequestLike
+from apimatic_core.adapters.flask_like import FlaskRequestLike
+from apimatic_core.adapters.starlette_like import StarletteRequestLike
+
+
+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()}
+
+async def to_unified_request(req: Union[StarletteRequestLike, FlaskRequestLike, DjangoRequestLike]) -> Request:
+    # --- Starlette/FastAPI ---
+    if isinstance(req, StarletteRequestLike):
+        headers = dict(req.headers)
+        raw = await req.body()
+        query = _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()
+        form: Dict[str, List[str]] = {}
+        if ct.startswith(("multipart/form-data", "application/x-www-form-urlencoded")):
+            form_data = await req.form()
+            for k in form_data.keys():
+                for v in form_data.getlist(k):
+                    if not (hasattr(v, "filename") and hasattr(v, "read")):
+                        form.setdefault(k, []).append(str(v))
+        return Request(method=req.method, path=path, url=url_str, headers=headers,
+                       raw_body=raw, query=query, cookies=cookies, form=form)
+
+    # --- Flask ---
+    if isinstance(req, FlaskRequestLike):
+        headers = dict(req.headers)
+        url_str: Optional[str] = getattr(req, "url", None)
+        path: str = req.path
+        raw: bytes = req.get_data(cache=True)
+        query = _as_listdict(req.args)
+        cookies = dict(req.cookies)
+        # best-effort cookie header fallback if the jar is empty
+        if not cookies:
+            cookie_header = headers.get("Cookie") or headers.get("cookie")
+            if cookie_header:
+                jar = SimpleCookie(); jar.load(cookie_header)
+                cookies = {k: morsel.value for k, morsel in jar.items()}
+        form = _as_listdict(req.form)
+        return Request(method=req.method, path=path, url=url_str, headers=headers,
+                       raw_body=raw, query=query, cookies=cookies, form=form)
+
+    # --- Django ---
+    if isinstance(req, DjangoRequestLike):
+        headers = dict(getattr(req, "headers", {}) or {})
+        # fallback for very old Django: META → headers
+        if not headers:
+            meta = getattr(req, "META", {}) or {}
+            headers = {k[5:].replace("_", "-"): str(v) for k, v in meta.items() if k.startswith("HTTP_")}
+        url_str = req.build_absolute_uri()
+        path = req.path
+        raw = bytes(getattr(req, "body", b"") or b"")
+        query = _as_listdict(getattr(req, "GET", {}))
+        cookies = dict(getattr(req, "COOKIES", {}) or {})
+        form = _as_listdict(getattr(req, "POST", {}))
+        return Request(method=req.method, path=path, url=url_str, headers=headers,
+                       raw_body=raw, query=query, cookies=cookies, form=form)
+
+    raise TypeError(f"Unsupported request type: {type(req)!r}")
+
+def _unwrap_local_proxy(obj):
+    """
+    Best-effort, dependency-free unwrapping for Flask's LocalProxy.
+
+    We intentionally DO NOT import `werkzeug.local.LocalProxy` to keep this adapter
+    framework-agnostic and avoid a hard dependency on Werkzeug/Flask internals.
+
+    Instead, we use *duck typing*: if the object exposes `_get_current_object()`
+    (the LocalProxy API), we call it to retrieve the real underlying request.
+    - If the object is not a LocalProxy, or unwrapping fails, we just return `obj`.
+    """
+    getter = getattr(obj, "_get_current_object", None)
+    if callable(getter):
+        try:
+            return getter()
+        except Exception:
+            # If unwrapping fails for any reason, fall back to the original object.
+            # The async core will still attempt structural dispatch via Protocols.
+            return obj
+    return obj
+
+def to_unified_request_sync(
+    req: Union[FlaskRequestLike, DjangoRequestLike]
+) -> Request:
+    """
+    Sync wrapper around the async `to_unified_request(...)` for use in
+    synchronous frameworks (Flask/Django) or sync code paths.
+
+    Key points:
+    - Accepts any object that *structurally* satisfies Flask/Django request Protocols.
+    - Transparently unwraps Flask's LocalProxy (without importing Werkzeug) using
+      duck-typing via `_get_current_object()`.
+    - Reuses the running event loop if present, otherwise creates one (common in
+      sync WSGI servers).
+    """
+    # 1) Unwrap LocalProxy-like objects (no werkzeug import; duck-typing)
+    req = _unwrap_local_proxy(req)
+
+    # 3) Bridge sync -> async: run the async core in (or with) an event loop.
+    try:
+        loop = asyncio.get_event_loop()
+    except RuntimeError:
+        # No running loop in this thread (typical in WSGI). Create one.
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+    # 4) Delegate to the single async entrypoint that handles Starlette/Flask/Django
+    #    via structural typing (Protocols).
+    return loop.run_until_complete(to_unified_request(req))
+
+async def from_starlette(req: StarletteRequestLike) -> Request:
+    return await to_unified_request(req)
+
+def from_flask(req: FlaskRequestLike) -> Request:
+    return to_unified_request_sync(req)
+
+def from_django(req: DjangoRequestLike) -> Request:
+    return to_unified_request_sync(req)

apimatic_core/adapters/starlette_like.py

@@ -0,0 +1,12 @@
+from typing import Any, Mapping, Coroutine
+from typing_extensions import Protocol, runtime_checkable
+
+@runtime_checkable
+class StarletteRequestLike(Protocol):
+    method: str
+    headers: Mapping[str, str]
+    cookies: Mapping[str, str]
+    query_params: Mapping[str, Any]
+    url: Any            # __str__ -> URL; has .path: str
+    def body(self) -> Coroutine[Any, Any, bytes]: ...
+    def form(self) -> Coroutine[Any, Any, Any]: ...

tests/apimatic_core/security/signature_verification/test_hmac_signature_verifier.py

@@ -34,7 +34,6 @@ def _get(name: str, default=None):
         cookies=dict(_get("cookies", {}) or {}),
         raw_body=_get("raw_body", None),
         form=dict(_get("form", {}) or {}),
-        files=dict(_get("files", {}) or {}),
     )
     # Apply overrides
     payload.update(overrides)
@@ -153,7 +152,6 @@ def req_base(self) -> Request:
             cookies={},   # Mapping[str, str]
             raw_body=b'{"event":{"id":"evt_1"},"payload":{"checksum":"abc"}}',
             form={},      # Mapping[str, List[str]]
-            files={},     # Mapping[str, List[FilePart]]
         )
 
     @pytest.fixture