feat: OData reusable transport

Author: NicoleMGomes

src/sap_cloud_sdk/core/odata/__init__.py

@@ -0,0 +1 @@
+"""Shared OData v4 abstractions for the SAP Cloud SDK."""

src/sap_cloud_sdk/core/odata/_async_transport.py

@@ -0,0 +1,218 @@
+"""Asynchronous HTTP transport for OData v4 services."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+import httpx
+
+from sap_cloud_sdk.core.odata.exceptions import (
+    ODataAuthError,
+    ODataCsrfError,
+    ODataNotFoundError,
+    ODataRequestError,
+)
+
+logger = logging.getLogger(__name__)
+
+_CSRF_HEADER = "X-CSRF-Token"
+_CSRF_FETCH_VALUE = "Fetch"
+_CSRF_FETCH_TIMEOUT = 10
+_REQUEST_TIMEOUT = 30
+
+
+class AsyncODataHttpTransport:
+    """Asynchronous HTTP transport for OData v4 services.
+
+    Mirrors :class:`~sap_cloud_sdk.core.odata._transport.ODataHttpTransport`
+    but uses ``httpx.AsyncClient``.  Use as an async context manager::
+
+        async with AsyncODataHttpTransport(base_url, client) as t:
+            data = await t.get("BusinessPartnerSet")
+
+    Args:
+        base_url: Root URL of the OData service.
+        client: Pre-configured ``httpx.AsyncClient``.
+        csrf_enabled: Whether to fetch and attach CSRF tokens on mutating
+            requests.
+    """
+
+    def __init__(
+        self,
+        base_url: str,
+        client: httpx.AsyncClient,
+        csrf_enabled: bool = True,
+    ) -> None:
+        self._base_url = base_url.rstrip("/")
+        self._client = client
+        self._csrf_enabled = csrf_enabled
+        self._csrf_token: str | None = None
+        self._csrf_lock = asyncio.Lock()
+
+    async def __aenter__(self) -> "AsyncODataHttpTransport":
+        return self
+
+    async def __aexit__(self, *args: Any) -> None:
+        await self._client.aclose()
+
+    # ------------------------------------------------------------------
+    # Public verbs
+    # ------------------------------------------------------------------
+
+    async def get(
+        self, path: str, params: dict[str, Any] | None = None
+    ) -> dict[str, Any]:
+        """Execute a GET and return the parsed JSON body."""
+        return await self._request("GET", path, params=params)
+
+    async def post(self, path: str, body: dict[str, Any]) -> dict[str, Any]:
+        """Execute a POST with CSRF and return the parsed JSON body."""
+        return await self._send_with_csrf("POST", path, json=body)
+
+    async def patch(
+        self,
+        path: str,
+        body: dict[str, Any],
+        etag: str | None = None,
+    ) -> dict[str, Any]:
+        extra: dict[str, str] = {}
+        if etag is not None:
+            extra["If-Match"] = etag
+        return await self._send_with_csrf("PATCH", path, json=body, extra_headers=extra)
+
+    async def put(
+        self,
+        path: str,
+        body: dict[str, Any],
+        etag: str | None = None,
+    ) -> dict[str, Any]:
+        extra: dict[str, str] = {}
+        if etag is not None:
+            extra["If-Match"] = etag
+        return await self._send_with_csrf("PUT", path, json=body, extra_headers=extra)
+
+    async def delete(self, path: str, etag: str | None = None) -> None:
+        extra: dict[str, str] = {}
+        if etag is not None:
+            extra["If-Match"] = etag
+        await self._send_with_csrf("DELETE", path, extra_headers=extra)
+
+    def absolute_url(self, path: str) -> str:
+        return self._base_url + "/" + path.lstrip("/")
+
+    # ------------------------------------------------------------------
+    # Internal
+    # ------------------------------------------------------------------
+
+    async def _send_with_csrf(
+        self,
+        method: str,
+        path: str,
+        *,
+        json: Any | None = None,
+        extra_headers: dict[str, str] | None = None,
+    ) -> dict[str, Any]:
+        csrf_headers: dict[str, str] = {}
+        if self._csrf_enabled:
+            csrf_headers[_CSRF_HEADER] = await self._get_csrf_token()
+
+        merged = {**(extra_headers or {}), **csrf_headers}
+        try:
+            return await self._request(method, path, json=json, extra_headers=merged)
+        except ODataAuthError as exc:
+            if exc.status_code == 403 and self._csrf_enabled:
+                await self._invalidate_csrf_token()
+                csrf_headers[_CSRF_HEADER] = await self._get_csrf_token()
+                merged = {**(extra_headers or {}), **csrf_headers}
+                return await self._request(method, path, json=json, extra_headers=merged)
+            raise
+
+    async def _get_csrf_token(self) -> str:
+        async with self._csrf_lock:
+            if self._csrf_token is not None:
+                return self._csrf_token
+
+        token = await self._fetch_csrf_token()
+        async with self._csrf_lock:
+            if self._csrf_token is None:
+                self._csrf_token = token
+            return self._csrf_token  # type: ignore[return-value]
+
+    async def _invalidate_csrf_token(self) -> None:
+        async with self._csrf_lock:
+            self._csrf_token = None
+
+    async def _fetch_csrf_token(self) -> str:
+        url = self._base_url + "/"
+        try:
+            resp = await self._client.get(
+                url,
+                headers={_CSRF_HEADER: _CSRF_FETCH_VALUE},
+                timeout=_CSRF_FETCH_TIMEOUT,
+            )
+        except httpx.RequestError as exc:
+            raise ODataCsrfError(f"Async CSRF fetch failed: {exc}") from exc
+
+        token = resp.headers.get(_CSRF_HEADER, "")
+        if not token:
+            raise ODataCsrfError(
+                f"Service did not return a CSRF token (HTTP {resp.status_code})"
+            )
+        return token
+
+    async def _request(
+        self,
+        method: str,
+        path: str,
+        *,
+        params: dict[str, Any] | None = None,
+        json: Any | None = None,
+        extra_headers: dict[str, str] | None = None,
+    ) -> dict[str, Any]:
+        url = self.absolute_url(path)
+        headers: dict[str, str] = {
+            "Accept": "application/json",
+            "Content-Type": "application/json",
+        }
+        if extra_headers:
+            headers.update(extra_headers)
+
+        logger.debug("%s %s params=%s", method, url, params)
+        try:
+            resp = await self._client.request(
+                method=method,
+                url=url,
+                headers=headers,
+                params=params,
+                json=json,
+                timeout=_REQUEST_TIMEOUT,
+            )
+        except httpx.RequestError as exc:
+            raise ODataCsrfError(f"Request failed: {exc}") from exc
+
+        self._raise_for_status(resp)
+
+        if resp.status_code == 204 or not resp.content:
+            return {}
+        return resp.json()
+
+    def _raise_for_status(self, response: httpx.Response) -> None:
+        if response.status_code == 404:
+            raise ODataNotFoundError(_HttpxResponseAdapter(response))
+        if response.status_code in (401, 403):
+            raise ODataAuthError(_HttpxResponseAdapter(response))
+        if not (200 <= response.status_code < 300):
+            raise ODataRequestError(_HttpxResponseAdapter(response))
+
+
+class _HttpxResponseAdapter:
+    """Minimal adapter so httpx.Response can be passed to ODataRequestError."""
+
+    def __init__(self, response: httpx.Response) -> None:
+        self.status_code = response.status_code
+        self._response = response
+
+    def json(self) -> Any:
+        return self._response.json()

src/sap_cloud_sdk/core/odata/_csrf.py

@@ -0,0 +1,71 @@
+"""CSRF token fetch-and-cache for OData v4 mutating requests."""
+
+from __future__ import annotations
+
+import threading
+from typing import TYPE_CHECKING
+
+import requests as _requests
+
+from sap_cloud_sdk.core.odata.exceptions import ODataCsrfError
+
+if TYPE_CHECKING:
+    from ._transport import ODataHttpTransport
+
+_CSRF_HEADER = "X-CSRF-Token"
+_CSRF_FETCH_VALUE = "Fetch"
+_FETCH_TIMEOUT = 10
+
+
+class CsrfTokenProvider:
+    """Fetch and cache a CSRF token for one OData service root.
+
+    The token is fetched lazily on the first mutating request and cached
+    until it is invalidated (typically after a 403 response).
+
+    Thread-safe: internal state is protected by a :class:`threading.Lock`.
+
+    Args:
+        transport: The owning :class:`ODataHttpTransport` whose session and
+            base URL are used to perform the CSRF-fetch GET.
+    """
+
+    def __init__(self, transport: "ODataHttpTransport") -> None:
+        self._transport = transport
+        self._token: str | None = None
+        self._lock = threading.Lock()
+
+    def get(self) -> str:
+        """Return the cached CSRF token, fetching from the service if needed."""
+        with self._lock:
+            if self._token is not None:
+                return self._token
+
+        token = self._fetch()
+        with self._lock:
+            if self._token is None:
+                self._token = token
+            return self._token
+
+    def invalidate(self) -> None:
+        """Discard the cached token so the next call re-fetches."""
+        with self._lock:
+            self._token = None
+
+    def _fetch(self) -> str:
+        url = self._transport._base_url + "/"
+        try:
+            resp = self._transport._session.get(
+                url,
+                headers={_CSRF_HEADER: _CSRF_FETCH_VALUE},
+                timeout=_FETCH_TIMEOUT,
+            )
+        except _requests.RequestException as exc:
+            raise ODataCsrfError(f"CSRF fetch failed: {exc}") from exc
+
+        token = resp.headers.get(_CSRF_HEADER, "")
+        if not token:
+            raise ODataCsrfError(
+                f"Service did not return a CSRF token (HTTP {resp.status_code})"
+            )
+        return token

src/sap_cloud_sdk/core/odata/_filter.py

@@ -0,0 +1,97 @@
+"""Filter expression DSL for OData v4 $filter query options."""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+def _format_value(value: Any) -> str:
+    """Serialise a Python value to an OData v4 literal."""
+    if isinstance(value, bool):
+        return "true" if value else "false"
+    if isinstance(value, str):
+        return "'" + value.replace("'", "''") + "'"
+    return str(value)
+
+
+class FilterExpression:
+    """Composable OData v4 ``$filter`` expression.
+
+    Build expressions via :meth:`field` and combine them with :meth:`and_`,
+    :meth:`or_`, and :meth:`not_`::
+
+        f = (
+            FilterExpression.field("Price").gt(100)
+            .and_(FilterExpression.field("Category").eq("Books"))
+        )
+        str(f)  # "(Price gt 100) and (Category eq 'Books')"
+    """
+
+    __slots__ = ("_expr",)
+
+    def __init__(self, expr: str) -> None:
+        self._expr = expr
+
+    @staticmethod
+    def field(name: str) -> "_FieldRef":
+        """Start a comparison expression for *name*."""
+        return _FieldRef(name)
+
+    def and_(self, other: "FilterExpression") -> "FilterExpression":
+        return FilterExpression(f"({self._expr}) and ({other._expr})")
+
+    def or_(self, other: "FilterExpression") -> "FilterExpression":
+        return FilterExpression(f"({self._expr}) or ({other._expr})")
+
+    def not_(self) -> "FilterExpression":
+        return FilterExpression(f"not ({self._expr})")
+
+    def __str__(self) -> str:
+        return self._expr
+
+    def __repr__(self) -> str:
+        return f"FilterExpression({self._expr!r})"
+
+    def __eq__(self, other: object) -> bool:
+        if isinstance(other, FilterExpression):
+            return self._expr == other._expr
+        return NotImplemented
+
+    def __hash__(self) -> int:
+        return hash(self._expr)
+
+
+class _FieldRef:
+    """Intermediate object: a field name awaiting a comparison operator."""
+
+    __slots__ = ("_name",)
+
+    def __init__(self, name: str) -> None:
+        self._name = name
+
+    def eq(self, value: Any) -> FilterExpression:
+        return FilterExpression(f"{self._name} eq {_format_value(value)}")
+
+    def ne(self, value: Any) -> FilterExpression:
+        return FilterExpression(f"{self._name} ne {_format_value(value)}")
+
+    def lt(self, value: Any) -> FilterExpression:
+        return FilterExpression(f"{self._name} lt {_format_value(value)}")
+
+    def le(self, value: Any) -> FilterExpression:
+        return FilterExpression(f"{self._name} le {_format_value(value)}")
+
+    def gt(self, value: Any) -> FilterExpression:
+        return FilterExpression(f"{self._name} gt {_format_value(value)}")
+
+    def ge(self, value: Any) -> FilterExpression:
+        return FilterExpression(f"{self._name} ge {_format_value(value)}")
+
+    def contains(self, value: str) -> FilterExpression:
+        return FilterExpression(f"contains({self._name}, {_format_value(value)})")
+
+    def starts_with(self, value: str) -> FilterExpression:
+        return FilterExpression(f"startswith({self._name}, {_format_value(value)})")
+
+    def ends_with(self, value: str) -> FilterExpression:
+        return FilterExpression(f"endswith({self._name}, {_format_value(value)})")