PTHMINT-94: Http transport

README.md

@@ -70,4 +70,4 @@ If you create a pull request to suggest an improvement, we'll send you some Mult
 ## Want to be part of the team?
 
 Are you a developer interested in working at MultiSafepay? Check out
-our [job openings](https://www.multisafepay.com/careers/#jobopenings) and feel free to get in touch!
+our [job openings](https://www.multisafepay.com/careers/#jobopenings) and feel free to get in touch!

examples/transport/httpx_transport.py

@@ -0,0 +1,80 @@
+"""Example: Injecting an httpx-based transport.
+
+This example demonstrates the recommended pattern for using a different HTTP
+client than requests, without the SDK having to maintain “official” adapters.
+
+Requirements
+------------
+- `pip install httpx`
+- `API_KEY` in the environment (optionally via a `.env` + python-dotenv)
+
+Notes
+-----
+- `httpx.Response` already exposes `status_code`, `headers`, `.json()` and
+    `.raise_for_status()`, so we don't need to adapt the response object.
+- We only adapt the *transport* (how requests are executed) to match the SDK's
+    `HTTPTransport.request(...)` contract.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any, Dict, Optional
+
+from dotenv import load_dotenv
+
+from multisafepay import Sdk
+
+
+class HttpxTransport:
+    """Minimal `HTTPTransport` implementation backed by httpx."""
+
+    def __init__(self, client: Optional[Any] = None) -> None:
+        import httpx
+
+        self._httpx = httpx
+        self.client = client or httpx.Client()
+
+    def request(
+        self,
+        method: str,
+        url: str,
+        headers: Optional[Dict[str, str]] = None,
+        params: Optional[Dict[str, Any]] = None,
+        data: Optional[Any] = None,
+        **kwargs: Any,
+    ) -> Any:
+        # The SDK may pass `data` as a dict; in httpx this should be sent as `json=`.
+        request_kwargs: Dict[str, Any] = {
+            "method": method,
+            "url": url,
+            "headers": headers,
+            "params": params,
+            **kwargs,
+        }
+
+        if data is not None:
+            if isinstance(data, (dict, list)):
+                request_kwargs["json"] = data
+            else:
+                request_kwargs["content"] = data
+
+        return self.client.request(**request_kwargs)
+
+    def close(self) -> None:
+        self.client.close()
+
+
+if __name__ == "__main__":
+    load_dotenv()
+    api_key = os.getenv("API_KEY")
+    if not api_key:
+        raise SystemExit("Missing API_KEY env var")
+
+    transport = HttpxTransport()
+    try:
+        sdk = Sdk(api_key=api_key, is_production=False, transport=transport)
+        gateways = sdk.get_gateway_manager().get_gateways().get_data()
+        print(gateways)
+    finally:
+        transport.close()

examples/transport/request_transport.py

@@ -0,0 +1,71 @@
+"""Example: Custom requests.Session transport.
+
+This example shows how to configure a real `requests.Session` (Retry + connection
+pooling) and inject it into the SDK via `RequestsTransport`.
+
+Requires the optional extra: `multisafepay[requests]`.
+"""
+
+import os
+
+from dotenv import load_dotenv
+from requests import Session
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
+
+from multisafepay import Sdk
+from multisafepay.transport import RequestsTransport
+
+
+def create_custom_session() -> Session:
+    """Create a custom requests Session with retry logic and connection pooling."""
+    retry_strategy = Retry(
+        total=3,
+        backoff_factor=1,
+        status_forcelist=[429, 500, 502, 503, 504],
+        allowed_methods=["GET", "POST", "PATCH", "DELETE"],
+    )
+
+    adapter = HTTPAdapter(
+        max_retries=retry_strategy,
+        pool_connections=10,
+        pool_maxsize=20,
+    )
+
+    session = Session()
+    session.mount("http://", adapter)
+    session.mount("https://", adapter)
+
+    # Note: requests doesn't support a global default timeout on Session.
+    # Timeouts must be passed per request (or implemented in a custom transport).
+    session.headers.update(
+        {
+            "User-Agent": "multisafepay-python-sdk-examples",
+            "Accept": "application/json",
+        }
+    )
+
+    return session
+
+
+# Load environment variables from a .env file
+load_dotenv()
+
+# Retrieve the API key from the environment variables
+API_KEY = os.getenv("API_KEY")
+
+
+if __name__ == "__main__":
+    # Create a custom requests Session and inject it into the SDK
+    custom_session = create_custom_session()
+    transport = RequestsTransport(session=custom_session)
+
+    multisafepay_sdk: Sdk = Sdk(API_KEY, False, transport)
+    gateway_manager = multisafepay_sdk.get_gateway_manager()
+
+    get_gateways_response = gateway_manager.get_gateways()
+    gateway_listing = get_gateways_response.get_data()
+
+    print(gateway_listing)
+
+    custom_session.close()

examples/transport/urllib3_transport.py

@@ -0,0 +1,102 @@
+"""Example: Injecting an urllib3-based transport (with response adaptation).
+
+Unlike requests/httpx, urllib3 is lower-level and its response object does NOT
+expose the typical high-level interface (`status_code`, `.json()`,
+`.raise_for_status()`).
+
+This is why the example shows the full pattern:
+- A `Transport` that executes the request via `urllib3.PoolManager`
+- A `ResponseAdapter` that translates urllib3 responses into what the SDK expects
+
+Requirements
+------------
+- `pip install urllib3`
+- `API_KEY` in the environment (optionally via a `.env` + python-dotenv)
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import dataclass
+from typing import Any, Dict, Optional
+
+from dotenv import load_dotenv
+
+from multisafepay import Sdk
+
+
+@dataclass
+class Urllib3ResponseAdapter:
+    """Adapter: urllib3.HTTPResponse -> interface expected by the SDK."""
+
+    response: Any
+
+    @property
+    def status_code(self) -> int:
+        return int(getattr(self.response, "status"))
+
+    @property
+    def headers(self) -> Dict[str, str]:
+        raw = getattr(self.response, "headers", {})
+        return {str(k): str(v) for k, v in dict(raw).items()}
+
+    def json(self) -> Any:
+        data = getattr(self.response, "data", b"")
+        if data is None:
+            return None
+        text = data if isinstance(data, str) else data.decode("utf-8")
+        return json.loads(text) if text else {}
+
+    def raise_for_status(self) -> None:
+        if self.status_code >= 400:
+            raise Exception(f"HTTP Error {self.status_code}")
+
+
+class Urllib3Transport:
+    """Minimal `HTTPTransport` implementation backed by urllib3."""
+
+    def __init__(self, pool_manager: Optional[Any] = None) -> None:
+        urllib3 = __import__("urllib3")
+        self._pool = pool_manager or urllib3.PoolManager()
+
+    def request(
+        self,
+        method: str,
+        url: str,
+        headers: Optional[Dict[str, str]] = None,
+        params: Optional[Dict[str, Any]] = None,  # noqa: ARG002
+        data: Optional[Any] = None,
+        **kwargs: Any,
+    ) -> Urllib3ResponseAdapter:
+        # The SDK already bakes query parameters into `url`; avoid re-applying `params`.
+        body: Optional[bytes]
+        if data is None:
+            body = None
+        elif isinstance(data, (bytes, bytearray)):
+            body = bytes(data)
+        elif isinstance(data, str):
+            body = data.encode("utf-8")
+        else:
+            body = json.dumps(data).encode("utf-8")
+
+        resp = self._pool.request(
+            method=method,
+            url=url,
+            body=body,
+            headers=headers,
+            **kwargs,
+        )
+        return Urllib3ResponseAdapter(resp)
+
+
+if __name__ == "__main__":
+    load_dotenv()
+    api_key = os.getenv("API_KEY")
+    if not api_key:
+        raise SystemExit("Missing API_KEY env var")
+
+    transport = Urllib3Transport()
+    sdk = Sdk(api_key=api_key, is_production=False, transport=transport)
+    gateways = sdk.get_gateway_manager().get_gateways().get_data()
+    print(gateways)