feat: add DestinationHttpClient to simplify calls to target systems

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "sap-cloud-sdk"
-version = "0.18.2"
+version = "0.18.3"
 description = "SAP Cloud SDK for Python"
 readme = "README.md"
 license = "Apache-2.0"

src/sap_cloud_sdk/destination/__init__.py

@@ -48,6 +48,7 @@
 )
 from sap_cloud_sdk.destination.config import load_from_env_or_mount, DestinationConfig
 from sap_cloud_sdk.destination._http import TokenProvider, DestinationHttp
+from sap_cloud_sdk.destination._destination_http_client import DestinationHttpClient
 from sap_cloud_sdk.destination.client import DestinationClient
 from sap_cloud_sdk.destination.fragment_client import FragmentClient
 from sap_cloud_sdk.destination.certificate_client import CertificateClient
@@ -235,6 +236,7 @@ def create_certificate_client(
     "LocalDevDestinationClient",
     "LocalDevFragmentClient",
     "LocalDevCertificateClient",
+    "DestinationHttpClient",
     # Exceptions
     "DestinationError",
     "ClientCreationError",

src/sap_cloud_sdk/destination/_destination_http_client.py

@@ -0,0 +1,69 @@
+"""HTTP client for calling the target system described by a Destination."""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Optional
+
+import requests
+from requests import Response
+
+from sap_cloud_sdk.destination._models import Destination, DestinationType
+
+
+class DestinationHttpClient:
+    """Wraps requests.Session to call the target system described by a Destination.
+
+    Pre-bakes headers derived from the destination — ERP headers (sap-client,
+    sap-language), URL.headers.* properties, and auth tokens.
+
+    Usage::
+
+        dest = client.get_destination("my-erp")
+        http = DestinationHttpClient(dest)
+        response = http.request("GET", "/api/resource")
+    """
+
+    def __init__(self, destination: Destination) -> None:
+        if destination.type not in (DestinationType.HTTP, "HTTP"):
+            raise ValueError(
+                f"DestinationHttpClient only supports HTTP destinations, got: {destination.type}"
+            )
+
+        self._destination = destination
+        self._session = requests.Session()
+        self._session.headers.update(destination.get_headers())
+        self._base_url = destination.url.rstrip("/") if destination.url else ""
+
+    def request(
+        self,
+        method: str,
+        path: str,
+        *,
+        params: Optional[Dict[str, Any]] = None,
+        json: Optional[Any] = None,
+        headers: Optional[Dict[str, str]] = None,
+        **kwargs: Any,
+    ) -> Response:
+        """Send an HTTP request to the target system.
+
+        Args:
+            method: HTTP verb (GET, POST, PUT, PATCH, DELETE).
+            path: Path relative to the destination URL.
+            params: Optional query parameters.
+            json: Optional JSON body.
+            headers: Optional additional headers merged on top of pre-baked ones.
+            **kwargs: Passed through to requests.Session.request.
+
+        Returns:
+            requests.Response from the target system.
+        """
+        url = f"{self._base_url}/{path.lstrip('/')}" if path else self._base_url
+        return self._session.request(
+            method=method.upper(),
+            url=url,
+            params=params,
+            json=json,
+            headers=headers,
+            **kwargs,
+        )
+

src/sap_cloud_sdk/destination/_models.py

@@ -366,6 +366,41 @@ def to_dict(self) -> Dict[str, Any]:
                     payload[k] = v
         return payload
 
+    def get_erp_headers(self) -> Dict[str, str]:
+        """Return SAP ERP-specific headers derived from destination properties (sap-client, sap-language).
+
+        Returns:
+            Headers to inject into requests to the target system.
+        """
+        headers: Dict[str, str] = {}
+        if "sap-client" in self.properties:
+            headers["sap-client"] = self.properties["sap-client"]
+        if "sap-language" in self.properties:
+            headers["sap-language"] = self.properties["sap-language"]
+        return headers
+
+    def get_headers(self) -> Dict[str, str]:
+        """Return HTTP headers derived from this destination (ERP headers, URL.headers.* properties, and auth tokens), each overriding the previous on conflicting keys.
+
+        Returns:
+            Headers ready to inject into requests to the target system.
+        """
+        headers: Dict[str, str] = {}
+        headers.update(self.get_erp_headers())
+
+        _PREFIX = "URL.headers."
+        for key, value in self.properties.items():
+            if key.startswith(_PREFIX):
+                headers[key[len(_PREFIX) :]] = value
+
+        for token in self.auth_tokens:
+            key = token.http_header.get("key")
+            value = token.http_header.get("value")
+            if key and value:
+                headers[key] = value
+
+        return headers
+
 
 @dataclass
 class AuthToken:

src/sap_cloud_sdk/destination/client.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import warnings
 from typing import List, Optional, Callable, TypeVar
 
 from sap_cloud_sdk.core.telemetry import Module, Operation, record_metrics
@@ -202,6 +203,9 @@ def get_instance_destination(
     ) -> Optional[Destination | TransparentProxyDestination]:
         """Get a destination from the service instance scope.
 
+        .. deprecated::
+            Use ``get_destination()`` instead, which automatically retrieves auth tokens via the v2 API.
+
         Args:
             name: Destination name.
             proxy_enabled: Whether to route the request through a transparent proxy (if configured).
@@ -213,6 +217,12 @@ def get_instance_destination(
         Raises:
             DestinationOperationError: If an HTTP error occurs or response parsing fails.
         """
+        warnings.warn(
+            "get_instance_destination() is deprecated. "
+            "Use get_destination() instead, which also includes automatic token retrieval.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         try:
             if self._should_use_proxy(proxy_enabled):
                 return TransparentProxyDestination.from_proxy(
@@ -237,6 +247,9 @@ def get_subaccount_destination(
     ) -> Optional[Destination | TransparentProxyDestination]:
         """Get a destination from the subaccount scope with an access strategy.
 
+        .. deprecated::
+            Use ``get_destination()`` instead, which automatically retrieves auth tokens via the v2 API.
+
         Access strategies:
             - SUBSCRIBER_ONLY: Fetch only from subscriber context (tenant required)
             - PROVIDER_ONLY: Fetch only from provider context (no tenant required)
@@ -257,6 +270,12 @@ def get_subaccount_destination(
             DestinationOperationError: If tenant is missing for subscriber access strategies,
                                        on HTTP errors, or response parsing failures.
         """
+        warnings.warn(
+            "get_subaccount_destination() is deprecated. "
+            "Use get_destination() instead, which also includes automatic token retrieval.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         try:
             if self._should_use_proxy(proxy_enabled) and self._transparent_proxy:
                 return TransparentProxyDestination.from_proxy(

src/sap_cloud_sdk/destination/user-guide.md

@@ -24,7 +24,7 @@ fragment_client = create_fragment_client(instance="default")
 certificate_client = create_certificate_client(instance="default")
 
 # Instance-level read
-dest = client.get_instance_destination("my-destination")
+dest = client.get_instance_destination("my-destination")  # deprecated: use get_destination()
 fragment = fragment_client.get_instance_fragment("my-fragment")
 cert = certificate_client.get_instance_certificate("my-cert")
 
@@ -39,12 +39,12 @@ fragments = fragment_client.list_instance_fragments(tenant="tenant-subdomain")
 certificates = certificate_client.list_instance_certificates(tenant="tenant-subdomain")
 
 # Subaccount-level read: provider only (no tenant required)
-dest = client.get_subaccount_destination("my-destination", access_strategy=AccessStrategy.PROVIDER_ONLY)
+dest = client.get_subaccount_destination("my-destination", access_strategy=AccessStrategy.PROVIDER_ONLY)  # deprecated: use get_destination()
 fragment = fragment_client.get_subaccount_fragment("my-fragment", access_strategy=AccessStrategy.PROVIDER_ONLY)
 cert = certificate_client.get_subaccount_certificate("my-cert", access_strategy=AccessStrategy.PROVIDER_ONLY)
 
 # Subaccount-level read: subscriber-first (tenant required), fallback to provider
-dest = client.get_subaccount_destination("my-destination", access_strategy=AccessStrategy.SUBSCRIBER_FIRST, tenant="tenant-subdomain")
+dest = client.get_subaccount_destination("my-destination", access_strategy=AccessStrategy.SUBSCRIBER_FIRST, tenant="tenant-subdomain")  # deprecated: use get_destination()
 fragment = fragment_client.get_subaccount_fragment("my-fragment", access_strategy=AccessStrategy.SUBSCRIBER_FIRST, tenant="tenant-subdomain")
 cert = certificate_client.get_subaccount_certificate("my-cert", access_strategy=AccessStrategy.SUBSCRIBER_FIRST, tenant="tenant-subdomain")
 
@@ -126,8 +126,8 @@ The client produced by `create_client()` exposes the following operations:
 ```python
 class DestinationClient:
     # V1 Admin API - Read operations for destinations
-    def get_instance_destination(self, name: str, proxy_enabled: Optional[bool] = None) -> Optional[Destination | TransparentProxyDestination]: ...
-    def get_subaccount_destination(self, name: str, access_strategy: AccessStrategy = AccessStrategy.SUBSCRIBER_FIRST, tenant: Optional[str] = None, proxy_enabled: Optional[bool] = None) -> Optional[Destination | TransparentProxyDestination]: ...
+    def get_instance_destination(self, name: str, proxy_enabled: Optional[bool] = None) -> Optional[Destination | TransparentProxyDestination]: ...  # deprecated: use get_destination()
+    def get_subaccount_destination(self, name: str, access_strategy: AccessStrategy = AccessStrategy.SUBSCRIBER_FIRST, tenant: Optional[str] = None, proxy_enabled: Optional[bool] = None) -> Optional[Destination | TransparentProxyDestination]: ...  # deprecated: use get_destination()
     def list_instance_destinations(self, tenant: Optional[str] = None, filter: Optional[ListOptions] = None) -> PagedResult[Destination]: ...
     def list_subaccount_destinations(self, access_strategy: AccessStrategy = AccessStrategy.SUBSCRIBER_FIRST, tenant: Optional[str] = None, filter: Optional[ListOptions] = None) -> PagedResult[Destination]: ...
 
@@ -218,6 +218,55 @@ class CertificateClient:
 - Certificate `content` should be base64-encoded. Supported certificate types include PEM, JKS, P12, etc.
 - The v2 consumption API returns tokens in the `auth_tokens` field with ready-to-use HTTP headers in `http_header` dict.
 
+## Calling Target Systems
+
+`DestinationHttpClient` wraps `requests.Session` to call the target system described by a destination. It injects headers automatically so you don't have to handle auth tokens, ERP headers, or custom destination properties manually.
+
+> **Note:** `DestinationHttpClient` requires a destination fetched via the v2 API (`get_destination()`), which returns pre-fetched auth tokens. It does not support destinations fetched with the deprecated v1 methods.
+
+### Basic Usage
+
+```python
+from sap_cloud_sdk.destination import create_client, DestinationHttpClient
+
+client = create_client(instance="default")
+dest = client.get_destination("my-erp")
+
+http = DestinationHttpClient(dest)
+response = http.request("GET", "/api/resource")
+```
+
+### What headers are pre-baked
+
+When `DestinationHttpClient` is constructed, it reads the destination and pre-bakes the following headers into every request:
+
+1. **ERP headers** — `sap-client` and `sap-language` from destination properties (if present)
+2. **`URL.headers.*` properties** — any destination property prefixed with `URL.headers.` becomes a header (e.g. `URL.headers.apiKey = secret` → `apiKey: secret`)
+3. **Auth tokens** — pre-fetched by BTP and returned in `dest.auth_tokens`; each token's `http_header` is injected directly (e.g. `Authorization: Bearer eyJ...`)
+
+Auth tokens take precedence over `URL.headers.*` properties if both set the same header key.
+
+### Per-request headers
+
+Pass `headers=` to add or override headers for a single request:
+
+```python
+response = http.request("GET", "/api/resource", headers={"X-Correlation-ID": "abc123"})
+```
+
+Per-request headers are merged on top of the pre-baked session headers.
+
+### Using `get_headers()` directly
+
+If you manage your own HTTP client, use `dest.get_headers()` to get all derived headers as a plain dict:
+
+```python
+import requests
+
+dest = client.get_destination("my-erp")
+response = requests.get(dest.url + "/api/resource", headers=dest.get_headers())
+```
+
 ## Transparent Proxy Support
 
 The destination client supports routing requests through a transparent proxy. This enables access to on-premise systems and private network resources through a proxy deployed in your Kubernetes cluster.

tests/destination/unit/test_destination_http_client.py

@@ -0,0 +1,106 @@
+"""Unit tests for DestinationHttpClient."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from sap_cloud_sdk.destination._destination_http_client import DestinationHttpClient
+from sap_cloud_sdk.destination._models import AuthToken, Destination
+
+
+def _dest(**kwargs) -> Destination:
+    base = {"Name": "test", "Type": "HTTP", "URL": "https://example.com"}
+    base.update(kwargs)
+    return Destination.from_dict(base)
+
+
+def _auth_token(key: str, value: str) -> AuthToken:
+    return AuthToken(type="Bearer", value="raw", http_header={"key": key, "value": value})
+
+
+class TestDestinationHttpClientInit:
+    def test_raises_for_non_http_destination(self):
+        dest = Destination.from_dict({"Name": "test", "Type": "RFC"})
+        with pytest.raises(ValueError, match="only supports HTTP destinations"):
+            DestinationHttpClient(dest)
+
+    def test_erp_headers_pre_baked(self):
+        dest = _dest(**{"sap-client": "100", "sap-language": "en"})
+        client = DestinationHttpClient(dest)
+        assert client._session.headers["sap-client"] == "100"
+        assert client._session.headers["sap-language"] == "en"
+
+    def test_no_erp_headers_when_properties_empty(self):
+        dest = _dest()
+        client = DestinationHttpClient(dest)
+        assert "sap-client" not in client._session.headers
+        assert "sap-language" not in client._session.headers
+
+    def test_auth_header_pre_baked_from_auth_tokens(self):
+        dest = _dest()
+        dest.auth_tokens = [_auth_token("Authorization", "Bearer eyJ123")]
+        client = DestinationHttpClient(dest)
+        assert client._session.headers["Authorization"] == "Bearer eyJ123"
+
+    def test_multiple_auth_tokens_all_injected(self):
+        dest = _dest()
+        dest.auth_tokens = [
+            _auth_token("Authorization", "Bearer eyJ123"),
+            _auth_token("x-sap-security-session", "mysession"),
+        ]
+        client = DestinationHttpClient(dest)
+        assert client._session.headers["Authorization"] == "Bearer eyJ123"
+        assert client._session.headers["x-sap-security-session"] == "mysession"
+
+    def test_error_token_with_empty_values_is_skipped(self):
+        dest = _dest()
+        dest.auth_tokens = [_auth_token("", "")]
+        client = DestinationHttpClient(dest)
+        assert "Authorization" not in client._session.headers
+
+    def test_no_auth_header_when_auth_tokens_empty(self):
+        dest = _dest()
+        client = DestinationHttpClient(dest)
+        assert "Authorization" not in client._session.headers
+
+    def test_url_headers_properties_pre_baked(self):
+        dest = _dest(**{"URL.headers.apiKey": "secret", "URL.headers.X-Tenant": "acme"})
+        client = DestinationHttpClient(dest)
+        assert client._session.headers["apiKey"] == "secret"
+        assert client._session.headers["X-Tenant"] == "acme"
+
+
+class TestDestinationHttpClientRequest:
+    def setup_method(self):
+        self.dest = _dest()
+        self.client = DestinationHttpClient(self.dest)
+        self.mock_response = MagicMock()
+
+    def test_constructs_full_url(self):
+        with patch.object(self.client._session, "request", return_value=self.mock_response) as mock_req:
+            self.client.request("GET", "/api/v1/users")
+            assert mock_req.call_args[1]["url"] == "https://example.com/api/v1/users"
+
+    def test_uppercases_method(self):
+        with patch.object(self.client._session, "request", return_value=self.mock_response) as mock_req:
+            self.client.request("get", "/resource")
+            assert mock_req.call_args[1]["method"] == "GET"
+
+    def test_passes_params(self):
+        with patch.object(self.client._session, "request", return_value=self.mock_response) as mock_req:
+            self.client.request("GET", "/resource", params={"$top": "10"})
+            assert mock_req.call_args[1]["params"] == {"$top": "10"}
+
+    def test_passes_json_body(self):
+        with patch.object(self.client._session, "request", return_value=self.mock_response) as mock_req:
+            self.client.request("POST", "/resource", json={"key": "value"})
+            assert mock_req.call_args[1]["json"] == {"key": "value"}
+
+    def test_passes_extra_headers(self):
+        with patch.object(self.client._session, "request", return_value=self.mock_response) as mock_req:
+            self.client.request("GET", "/resource", headers={"X-Custom": "yes"})
+            assert mock_req.call_args[1]["headers"] == {"X-Custom": "yes"}
+
+    def test_returns_response(self):
+        with patch.object(self.client._session, "request", return_value=self.mock_response):
+            assert self.client.request("GET", "/resource") is self.mock_response