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,83 @@
+# adapter.py
+from typing import Dict, List, Optional, Union
+from http.cookies import SimpleCookie
+from apimatic_core_interfaces.http.request import Request, _as_listdict
+
+from apimatic_core.adapters.django_like import DjangoRequestLike
+from apimatic_core.adapters.flask_like import FlaskRequestLike
+from apimatic_core.adapters.starlette_like import StarletteRequestLike
+
+
+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")):
+            formdata = await req.form()
+            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))
+        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}")
+
+# Optional convenience wrappers (sync for Flask/Django hosts)
+def to_unified_request_sync(req: Union[FlaskRequestLike, DjangoRequestLike]) -> Request:
+    import asyncio
+    try:
+        loop = asyncio.get_event_loop()
+    except RuntimeError:
+        loop = asyncio.new_event_loop(); asyncio.set_event_loop(loop)
+    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]: ...