feat(dpi_ng/consent): add Data Privacy Integration NextGen Consent service integration to sdk

.env_integration_tests.example

@@ -57,6 +57,24 @@ CLOUD_SDK_CFG_DESTINATION_DEFAULT_TENANT_SUBDOMAIN=your-subscriber-tenant-subdom
 CLOUD_SDK_CFG_SDM_DEFAULT_URI=https://your-sdm-api-uri-here
 CLOUD_SDK_CFG_SDM_DEFAULT_UAA='{"url":"https://your-auth-url","clientid":"your-client-id","clientsecret":"your-client-secret","identityzone":"your-identity-zone"}'
 
+# DPI NG - CONSENT
+# Required for all consent integration tests
+CLOUD_SDK_CFG_DPI_NG_DEFAULT_BASE_URL=https://your-dpi-ng-service-host
+
+# Pick ONE of the three auth methods below:
+
+# Option A: Static Bearer Token
+# CLOUD_SDK_CFG_DPI_NG_CONSENT_DEFAULT_BEARER_TOKEN=your-bearer-token-here
+# Option B: OAuth 2.0 Client Credentials
+# CLOUD_SDK_CFG_DPI_NG_CONSENT_DEFAULT_TOKEN_URL=https://your-auth-host/oauth/token
+# CLOUD_SDK_CFG_DPI_NG_CONSENT_DEFAULT_CLIENT_ID=your-client-id
+# CLOUD_SDK_CFG_DPI_NG_CONSENT_DEFAULT_CLIENT_SECRET=your-client-secret
+# Option C: Mutual TLS (mTLS)
+CLOUD_SDK_CFG_DPI_NG_CONSENT_DEFAULT_CERT_FILE=/path/to/client.crt
+CLOUD_SDK_CFG_DPI_NG_CONSENT_DEFAULT_KEY_FILE=/path/to/client.key
+CLOUD_SDK_CFG_DPI_NG_CONSENT_DEFAULT_CA_FILE=/path/to/ca.crt   # optional
+CLOUD_SDK_CFG_DPI_NG_CONSENT_DEFAULT_TENANT_ID=your-tenant-id-here
+
 # OBJECT STORE
 CLOUD_SDK_CFG_OBJECTSTORE_DEFAULT_HOST=your-objectstore-host-here
 CLOUD_SDK_CFG_OBJECTSTORE_DEFAULT_ACCESS_KEY_ID=your-access-key-id-here

README.md

@@ -4,7 +4,7 @@
 
 ## About this project
 
-This SDK provides consistent interfaces for interacting with foundational services such as object storage, destination management, audit logging, data anonymization, telemetry, and secure credential handling.
+This SDK provides consistent interfaces for interacting with foundational services such as object storage, destination management, audit logging, data anonymization, data privacy integration, telemetry, and secure credential handling.
 
 The Python SDK offers a clean, type-safe API following Python best practices while maintaining compatibility with the SAP Application Foundation ecosystem.
 
@@ -24,6 +24,7 @@ The Python SDK offers a clean, type-safe API following Python best practices whi
 - **Secret Resolver**
 - **Telemetry & Observability**
 - **Data Anonymization Service**
+- **Data Privacy Integration Service**
 - **Print Service**
 
 ## Requirements and Setup
@@ -80,6 +81,7 @@ Each module has comprehensive usage guides:
 - [Telemetry](src/sap_cloud_sdk/core/telemetry/user-guide.md)
 - [Print](src/sap_cloud_sdk/print/user-guide.md)
 - [Data Anonymization](src/sap_cloud_sdk/core/data_anonymization/user-guide.md)
+- [Data Privacy Integration](src/sap_cloud_sdk/core/dpi_ng/consent/user-guide.md)
 
 ## Support, Feedback, Contributing
 

docs/INTEGRATION_TESTS.md

@@ -141,6 +141,42 @@ CLOUD_SDK_CFG_SDM_DEFAULT_UAA='{"url":"https://your-auth-url","clientid":"your-c
 
 **Note**: The test fixture automatically onboards test repositories (standard and version-enabled) at session start and cleans them up on teardown. No pre-existing repositories are required.
 
+### DPI NG Consent Integration Tests
+
+For DPI NG Consent integration tests, configure the following variables in `.env_integration_tests`:
+
+```bash
+# DPI NG Consent Configuration - use the shared base URL (applies to all DPI NG sub-services)
+CLOUD_SDK_CFG_DPI_NG_DEFAULT_BASE_URL=https://your-dpi-ng-service-host
+```
+
+Authentication is configurable via one of three methods:
+
+**Option A - Static Bearer Token**
+
+```bash
+CLOUD_SDK_CFG_DPI_NG_CONSENT_DEFAULT_BEARER_TOKEN=your-bearer-token-here
+```
+
+**Option B - OAuth 2.0 Client Credentials**
+
+```bash
+CLOUD_SDK_CFG_DPI_NG_CONSENT_DEFAULT_TOKEN_URL=https://your-auth-host/oauth/token
+CLOUD_SDK_CFG_DPI_NG_CONSENT_DEFAULT_CLIENT_ID=your-client-id
+CLOUD_SDK_CFG_DPI_NG_CONSENT_DEFAULT_CLIENT_SECRET=your-client-secret
+```
+
+**Option C - Mutual TLS (mTLS)**
+
+```bash
+CLOUD_SDK_CFG_DPI_NG_CONSENT_DEFAULT_CERT_FILE=/path/to/client.crt
+CLOUD_SDK_CFG_DPI_NG_CONSENT_DEFAULT_KEY_FILE=/path/to/client.key
+CLOUD_SDK_CFG_DPI_NG_CONSENT_DEFAULT_CA_FILE=/path/to/ca.crt   # optional
+CLOUD_SDK_CFG_DPI_NG_CONSENT_DEFAULT_TENANT_ID=your-tenant-id-here
+```
+
+Tests are skipped automatically when `CLOUD_SDK_CFG_DPI_NG_DEFAULT_BASE_URL` or all auth variables are missing.
+
 ### ObjectStore Integration Tests
 
 For ObjectStore integration tests, configure the following variables in `.env_integration_tests`:
@@ -168,6 +204,7 @@ uv run pytest tests/core/integration/auditlog -v
 uv run pytest tests/core/integration/data_anonymization -v
 uv run pytest tests/destination/integration/ -v
 uv run pytest tests/dms/integration/ -v
+uv run pytest tests/core/integration/dpi_ng/consent/ -v
 uv run pytest tests/objectstore/integration/ -v
 ```
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "sap-cloud-sdk"
-version = "0.37.0"
+version = "0.38.0"
 description = "SAP Cloud SDK for Python"
 readme = "README.md"
 license = "Apache-2.0"
@@ -27,6 +27,7 @@ dependencies = [
     "opentelemetry-api>=1.42.1",
     "opentelemetry-sdk>=1.42.1",
     "mcp>=1.1.0",
+    "python-odata>=0.7.0",
 ]
 
 [project.optional-dependencies]
@@ -61,6 +62,7 @@ dev = [
     "langchain-community>=0.3.0",
     "litellm>=1.40.0",
     "langchain-litellm>=0.6.6",
+    "responses>=0.25",
 ]
 
 [tool.pytest.ini_options]

src/sap_cloud_sdk/core/dpi_ng/__init__.py

@@ -0,0 +1,49 @@
+"""DPI Next Gen SDK modules.
+
+Shared building blocks for all DPI NG capabilities:
+
+- :mod:`sap_cloud_sdk.core.dpi_ng.auth`         — AuthProvider ABC + strategies
+- :mod:`sap_cloud_sdk.core.dpi_ng.config`        — BaseCapabilityConfig
+- :mod:`sap_cloud_sdk.core.dpi_ng.exceptions`    — DPINGError hierarchy
+- :mod:`sap_cloud_sdk.core.dpi_ng.odata_client`  — BaseODataClient
+"""
+
+from .auth import (
+    AuthProvider,
+    BearerTokenAuth,
+    ClientCertificateAuth,
+    ClientCredentialsAuth,
+)
+from .config import BaseCapabilityConfig
+from .exceptions import (
+    AuthenticationError,
+    AuthorizationError,
+    ClientCreationError,
+    ConflictError,
+    DPINGError,
+    NotFoundError,
+    ODataError,
+    ValidationError,
+)
+from .odata_client import BaseODataClient
+
+__all__ = [
+    # auth
+    "AuthProvider",
+    "BearerTokenAuth",
+    "ClientCredentialsAuth",
+    "ClientCertificateAuth",
+    # config
+    "BaseCapabilityConfig",
+    # exceptions
+    "DPINGError",
+    "ClientCreationError",
+    "AuthenticationError",
+    "AuthorizationError",
+    "ValidationError",
+    "NotFoundError",
+    "ConflictError",
+    "ODataError",
+    # odata transport
+    "BaseODataClient",
+]

src/sap_cloud_sdk/core/dpi_ng/auth.py

@@ -0,0 +1,213 @@
+"""Authentication strategy implementations shared across all DPI NG capabilities.
+
+Each provider implements AuthProvider.apply(), which configures the
+requests.Session passed to it to inject the chosen auth mechanism.
+
+Supported strategies:
+  - BearerTokenAuth          - static bearer token (e.g. already-fetched XSUAA token)
+  - ClientCredentialsAuth    - OAuth2 client_credentials flow with automatic token refresh
+  - ClientCertificateAuth    - mTLS with client certificate + private key
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from abc import ABC, abstractmethod
+
+import requests
+import requests.auth
+
+logger = logging.getLogger(__name__)
+
+
+class AuthProvider(ABC):
+    """Abstract base for all authentication strategies.
+
+    Subclasses configure a ``requests.Session`` to inject their chosen
+    mechanism (headers, cert, custom auth flow, etc.).
+    Adding a new auth type means subclassing this; nothing else changes.
+    """
+
+    @abstractmethod
+    def apply(self, session: requests.Session) -> None:
+        """Configure the requests.Session to inject authentication."""
+
+
+class BearerTokenAuth(AuthProvider):
+    """Static bearer token - use when the caller manages token lifecycle externally."""
+
+    def __init__(self, token: str) -> None:
+        """Store the bearer token.
+
+        Args:
+            token: A valid bearer token string (e.g. an already-fetched XSUAA access token).
+
+        Raises:
+            ValueError: If *token* is an empty string.
+        """
+        logger.info("Invoked BearerTokenAuth.__init__")
+        if not token:
+            logger.error("token is empty")
+            raise ValueError("token must not be empty")
+        self._token = token
+        logger.info("Exiting BearerTokenAuth.__init__")
+
+    def apply(self, session: requests.Session) -> None:
+        """Set the ``Authorization: Bearer <token>`` header on the session.
+
+        Args:
+            session: The ``requests.Session`` to configure.
+        """
+        logger.info("Invoked BearerTokenAuth.apply")
+        session.headers["Authorization"] = f"Bearer {self._token}"
+        logger.info("Exiting BearerTokenAuth.apply")
+
+
+class ClientCredentialsAuth(AuthProvider):
+    """OAuth2 client_credentials flow with automatic token refresh.
+
+    Fetches a bearer token from ``token_url`` using ``client_id`` / ``client_secret``
+    and refreshes it transparently 60 seconds before it expires.
+    """
+
+    def __init__(self, token_url: str, client_id: str, client_secret: str) -> None:
+        """Store OAuth2 credentials for lazy token fetching.
+
+        Args:
+            token_url: Full URL of the OAuth2 token endpoint
+                (e.g. ``https://<subdomain>.authentication.eu10.hana.ondemand.com/oauth/token``).
+            client_id: OAuth2 client identifier.
+            client_secret: OAuth2 client secret.
+
+        Raises:
+            ValueError: If any of *token_url*, *client_id*, or *client_secret* is empty.
+        """
+        logger.info("Invoked ClientCredentialsAuth.__init__")
+        if not token_url or not client_id or not client_secret:
+            logger.error("token_url, client_id, or client_secret is empty")
+            raise ValueError("token_url, client_id, and client_secret are all required")
+        self._token_url = token_url
+        self._client_id = client_id
+        self._client_secret = client_secret
+        logger.info("Exiting ClientCredentialsAuth.__init__")
+
+    def apply(self, session: requests.Session) -> None:
+        """Attach the OAuth2 flow handler to the session.
+
+        Tokens are fetched on the first request and refreshed automatically
+        60 seconds before expiry.
+
+        Args:
+            session: The ``requests.Session`` to configure.
+        """
+        logger.info("Invoked ClientCredentialsAuth.apply")
+        session.auth = _OAuth2Flow(
+            self._token_url, self._client_id, self._client_secret
+        )
+        logger.info("Exiting ClientCredentialsAuth.apply")
+
+
+class ClientCertificateAuth(AuthProvider):
+    """Mutual TLS (mTLS) authentication using a client certificate and private key.
+
+    Args:
+        cert_file: Path to the client certificate PEM file.
+        key_file: Path to the client private key PEM file.
+        ca_file: Path to the CA certificate PEM file for server verification.
+                 When omitted, the system CA bundle is used.
+    """
+
+    def __init__(
+        self,
+        cert_file: str,
+        key_file: str,
+        ca_file: str | None = None,
+    ) -> None:
+        """Store mTLS file paths.
+
+        Args:
+            cert_file: Path to the PEM-encoded client certificate file.
+            key_file: Path to the PEM-encoded client private key file.
+            ca_file: Path to a PEM-encoded CA certificate file for server
+                verification. When omitted the system CA bundle is used.
+
+        Raises:
+            ValueError: If *cert_file* or *key_file* is empty.
+        """
+        logger.info("Invoked ClientCertificateAuth.__init__")
+        if not cert_file or not key_file:
+            logger.error("cert_file or key_file is empty")
+            raise ValueError("cert_file and key_file are required")
+        self._cert_file = cert_file
+        self._key_file = key_file
+        self._ca_file = ca_file
+        logger.info("Exiting ClientCertificateAuth.__init__")
+
+    def apply(self, session: requests.Session) -> None:
+        """Configure the session with the client cert/key pair and optional CA bundle.
+
+        Args:
+            session: The ``requests.Session`` to configure.
+        """
+        logger.info("Invoked ClientCertificateAuth.apply")
+        session.cert = (self._cert_file, self._key_file)
+        if self._ca_file:
+            session.verify = self._ca_file  # ty: ignore[invalid-assignment]
+            logger.debug("Custom CA bundle applied — ca_file=%s", self._ca_file)
+        logger.info("Exiting ClientCertificateAuth.apply")
+
+
+# ------------------------------------------------------------------
+# Internal OAuth2 flow - not part of the public API
+# ------------------------------------------------------------------
+
+
+class _OAuth2Flow(requests.auth.AuthBase):
+    """requests.auth.AuthBase implementation that handles token fetch and refresh."""
+
+    # 60-second buffer before actual expiry to avoid clock-skew races
+    _EXPIRY_BUFFER = 60.0
+
+    def __init__(self, token_url: str, client_id: str, client_secret: str) -> None:
+        """Initialise with OAuth2 endpoint and credentials; token is fetched lazily."""
+        self._token_url = token_url
+        self._client_id = client_id
+        self._client_secret = client_secret
+        self._access_token: str | None = None
+        self._expires_at: float = 0.0
+
+    def __call__(self, r: requests.PreparedRequest) -> requests.PreparedRequest:
+        """Inject a valid bearer token, refreshing it if expired or not yet fetched."""
+        if self._is_expired():
+            logger.debug("Token expired or absent — fetching new token")
+            self._fetch_token()
+        r.headers["Authorization"] = f"Bearer {self._access_token}"  # ty: ignore[invalid-assignment]
+        return r
+
+    def _is_expired(self) -> bool:
+        """Return True if no token has been fetched or the expiry buffer has been reached."""
+        return self._access_token is None or time.monotonic() >= self._expires_at
+
+    def _fetch_token(self) -> None:
+        """POST to the token URL and store the new access token and its expiry time."""
+        logger.info("Invoked _OAuth2Flow._fetch_token")
+        try:
+            resp = requests.post(
+                self._token_url,
+                data={
+                    "grant_type": "client_credentials",
+                    "client_id": self._client_id,
+                    "client_secret": self._client_secret,
+                },
+            )
+            resp.raise_for_status()
+            payload = resp.json()
+            self._access_token = payload["access_token"]
+            expires_in: float = float(payload.get("expires_in", 3600))
+            self._expires_at = time.monotonic() + expires_in - self._EXPIRY_BUFFER
+            logger.debug("Token acquired — expires_in=%.0fs", expires_in)
+        except Exception:
+            logger.exception("Failed to fetch access token from %s", self._token_url)
+            raise
+        logger.info("Exiting _OAuth2Flow._fetch_token")