Skip to content

Latest commit

 

History

History
974 lines (797 loc) · 38.1 KB

File metadata and controls

974 lines (797 loc) · 38.1 KB

Destination User Guide

This module integrates with SAP BTP Destination Service to manage destinations, fragments, and certificates at subaccount and service instance levels. It uses a Pythonic dataclass pattern for type-safe message construction.

Installation

This package is part of the SAP Cloud SDK for Python. Import and use it directly in your application.

Quick Start

from sap_cloud_sdk.destination import (
    create_client,
    create_fragment_client,
    create_certificate_client,
    Level,
    AccessStrategy,
    ConsumptionLevel,
    ConsumptionOptions,
)

client = create_client(instance="default")
fragment_client = create_fragment_client(instance="default")
certificate_client = create_certificate_client(instance="default")

# Instance-level read
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")

# Instance-level list: provider context (no tenant)
destinations = client.list_instance_destinations()
fragments = fragment_client.list_instance_fragments()
certificates = certificate_client.list_instance_certificates()

# Instance-level list: subscriber context (tenant provided)
destinations = client.list_instance_destinations(tenant="tenant-subdomain")
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
)  # 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",
)  # 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",
)

# Fragment write operations with tenant (subscriber context)
new_fragment = Fragment(
    name="my-fragment", properties={"URL": "https://api.example.com"}
)
fragment_client.create_fragment(
    new_fragment, level=Level.SUB_ACCOUNT, tenant="tenant-subdomain"
)
fragment_client.update_fragment(
    new_fragment, level=Level.SUB_ACCOUNT, tenant="tenant-subdomain"
)
fragment_client.delete_fragment(
    "my-fragment", level=Level.SUB_ACCOUNT, tenant="tenant-subdomain"
)

# Fragment write operations without tenant (provider context)
fragment_client.create_fragment(new_fragment, level=Level.SUB_ACCOUNT)
fragment_client.update_fragment(new_fragment, level=Level.SUB_ACCOUNT)
fragment_client.delete_fragment("my-fragment", level=Level.SUB_ACCOUNT)

# Destination write operations with tenant (subscriber context)
new_dest = Destination(name="my-dest", type="HTTP", url="https://api.example.com")
client.create_destination(new_dest, level=Level.SUB_ACCOUNT, tenant="tenant-subdomain")
client.update_destination(new_dest, level=Level.SUB_ACCOUNT, tenant="tenant-subdomain")
client.delete_destination("my-dest", level=Level.SUB_ACCOUNT, tenant="tenant-subdomain")

# Destination write operations without tenant (provider context)
client.create_destination(new_dest, level=Level.SUB_ACCOUNT)
client.update_destination(new_dest, level=Level.SUB_ACCOUNT)
client.delete_destination("my-dest", level=Level.SUB_ACCOUNT)

# Certificate write operations with tenant (subscriber context)
from sap_cloud_sdk.destination import create_certificate_client
from sap_cloud_sdk.destination._models import Certificate

new_cert = Certificate(name="my-cert.pem", content="base64-encoded-content", type="PEM")
certificate_client.create_certificate(
    new_cert, level=Level.SUB_ACCOUNT, tenant="tenant-subdomain"
)
certificate_client.update_certificate(
    new_cert, level=Level.SUB_ACCOUNT, tenant="tenant-subdomain"
)
certificate_client.delete_certificate(
    "my-cert.pem", level=Level.SUB_ACCOUNT, tenant="tenant-subdomain"
)

# Certificate write operations without tenant (provider context)
certificate_client.create_certificate(new_cert, level=Level.SUB_ACCOUNT)
certificate_client.update_certificate(new_cert, level=Level.SUB_ACCOUNT)
certificate_client.delete_certificate("my-cert.pem", level=Level.SUB_ACCOUNT)

# Label management with tenant (subscriber context)
from sap_cloud_sdk.destination import Label, PatchLabels

labels = [Label(key="env", values=["prod", "eu"])]

client.update_destination_labels(
    "my-dest", labels, level=Level.SUB_ACCOUNT, tenant="tenant-subdomain"
)
client.patch_destination_labels(
    "my-dest",
    PatchLabels(action="ADD", labels=labels),
    level=Level.SUB_ACCOUNT,
    tenant="tenant-subdomain",
)
retrieved = client.get_destination_labels(
    "my-dest", level=Level.SUB_ACCOUNT, tenant="tenant-subdomain"
)

fragment_client.update_fragment_labels(
    "my-fragment", labels, level=Level.SUB_ACCOUNT, tenant="tenant-subdomain"
)
certificate_client.get_certificate_labels(
    "my-cert.pem", level=Level.SUB_ACCOUNT, tenant="tenant-subdomain"
)

# Label management without tenant (provider context)
client.update_destination_labels("my-dest", labels, level=Level.SUB_ACCOUNT)
client.patch_destination_labels(
    "my-dest", PatchLabels(action="DELETE", labels=labels), level=Level.SUB_ACCOUNT
)

Concepts

  • Level (v1 admin API — write operations, label operations):

    • SERVICE_INSTANCE: Operates on instance destinations
    • SUB_ACCOUNT: Operates on subaccount destinations
  • ConsumptionLevel (v2 consumption API — get_destination only):

    • PROVIDER_SUBACCOUNT: Provider subaccount scope
    • PROVIDER_INSTANCE: Provider service instance scope
    • SUBACCOUNT: Subscriber subaccount scope
    • INSTANCE: Subscriber service instance scope
  • AccessStrategy (applies to subaccount reads):

    • SUBSCRIBER_ONLY: Only subscriber (tenant required)
    • PROVIDER_ONLY: Only provider (no tenant)
    • SUBSCRIBER_FIRST: Try subscriber, fallback to provider (tenant required)
    • PROVIDER_FIRST: Try provider, fallback to subscriber (tenant required)

API

Destination Client

The client produced by create_client() exposes the following operations:

class DestinationClient:
    # V1 Admin API - Read operations for destinations
    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]: ...

    # V1 Admin API - Write operations
    def create_destination(
        self,
        dest: Destination,
        level: Optional[Level] = Level.SUB_ACCOUNT,
        tenant: Optional[str] = None,
    ) -> None: ...
    def update_destination(
        self,
        dest: Destination,
        level: Optional[Level] = Level.SUB_ACCOUNT,
        tenant: Optional[str] = None,
    ) -> None: ...
    def delete_destination(
        self,
        name: str,
        level: Optional[Level] = Level.SUB_ACCOUNT,
        tenant: Optional[str] = None,
    ) -> None: ...
    def get_destination_labels(
        self,
        name: str,
        level: Optional[Level] = Level.SUB_ACCOUNT,
        tenant: Optional[str] = None,
    ) -> List[Label]: ...
    def update_destination_labels(
        self,
        name: str,
        labels: List[Label],
        level: Optional[Level] = Level.SUB_ACCOUNT,
        tenant: Optional[str] = None,
    ) -> None: ...
    def patch_destination_labels(
        self,
        name: str,
        patch: PatchLabels,
        level: Optional[Level] = Level.SUB_ACCOUNT,
        tenant: Optional[str] = None,
    ) -> None: ...

    # V2 Runtime API - Destination consumption with automatic token retrieval
    def get_destination(
        self,
        name: str,
        level: Optional[ConsumptionLevel] = None,
        options: Optional[ConsumptionOptions] = None,
        proxy_enabled: Optional[bool] = None,
        tenant: Optional[str] = None,
    ) -> Optional[Destination | TransparentProxyDestination]: ...

Fragment Client

The fragment client produced by create_fragment_client() exposes the following operations:

class FragmentClient:
    def get_instance_fragment(self, name: str) -> Optional[Fragment]: ...
    def get_subaccount_fragment(
        self,
        name: str,
        access_strategy: AccessStrategy = AccessStrategy.SUBSCRIBER_FIRST,
        tenant: Optional[str] = None,
    ) -> Optional[Fragment]: ...
    def list_instance_fragments(
        self, tenant: Optional[str] = None, filter: Optional[ListOptions] = None
    ) -> List[Fragment]: ...
    def list_subaccount_fragments(
        self,
        access_strategy: AccessStrategy = AccessStrategy.SUBSCRIBER_FIRST,
        tenant: Optional[str] = None,
        filter: Optional[ListOptions] = None,
    ) -> List[Fragment]: ...
    def create_fragment(
        self,
        fragment: Fragment,
        level: Optional[Level] = Level.SUB_ACCOUNT,
        tenant: Optional[str] = None,
    ) -> None: ...
    def update_fragment(
        self,
        fragment: Fragment,
        level: Optional[Level] = Level.SUB_ACCOUNT,
        tenant: Optional[str] = None,
    ) -> None: ...
    def delete_fragment(
        self,
        name: str,
        level: Optional[Level] = Level.SUB_ACCOUNT,
        tenant: Optional[str] = None,
    ) -> None: ...
    def get_fragment_labels(
        self,
        name: str,
        level: Optional[Level] = Level.SUB_ACCOUNT,
        tenant: Optional[str] = None,
    ) -> List[Label]: ...
    def update_fragment_labels(
        self,
        name: str,
        labels: List[Label],
        level: Optional[Level] = Level.SUB_ACCOUNT,
        tenant: Optional[str] = None,
    ) -> None: ...
    def patch_fragment_labels(
        self,
        name: str,
        patch: PatchLabels,
        level: Optional[Level] = Level.SUB_ACCOUNT,
        tenant: Optional[str] = None,
    ) -> None: ...

Certificate Client

The certificate client produced by create_certificate_client() exposes the following operations:

class CertificateClient:
    def get_instance_certificate(self, name: str) -> Optional[Certificate]: ...
    def get_subaccount_certificate(
        self,
        name: str,
        access_strategy: AccessStrategy = AccessStrategy.SUBSCRIBER_FIRST,
        tenant: Optional[str] = None,
    ) -> Optional[Certificate]: ...
    def list_instance_certificates(
        self, tenant: Optional[str] = None, filter: Optional[ListOptions] = None
    ) -> PagedResult[Certificate]: ...
    def list_subaccount_certificates(
        self,
        access_strategy: AccessStrategy = AccessStrategy.SUBSCRIBER_FIRST,
        tenant: Optional[str] = None,
        filter: Optional[ListOptions] = None,
    ) -> PagedResult[Certificate]: ...
    def create_certificate(
        self,
        certificate: Certificate,
        level: Optional[Level] = Level.SUB_ACCOUNT,
        tenant: Optional[str] = None,
    ) -> None: ...
    def update_certificate(
        self,
        certificate: Certificate,
        level: Optional[Level] = Level.SUB_ACCOUNT,
        tenant: Optional[str] = None,
    ) -> None: ...
    def delete_certificate(
        self,
        name: str,
        level: Optional[Level] = Level.SUB_ACCOUNT,
        tenant: Optional[str] = None,
    ) -> None: ...
    def get_certificate_labels(
        self,
        name: str,
        level: Optional[Level] = Level.SUB_ACCOUNT,
        tenant: Optional[str] = None,
    ) -> List[Label]: ...
    def update_certificate_labels(
        self,
        name: str,
        labels: List[Label],
        level: Optional[Level] = Level.SUB_ACCOUNT,
        tenant: Optional[str] = None,
    ) -> None: ...
    def patch_certificate_labels(
        self,
        name: str,
        patch: PatchLabels,
        level: Optional[Level] = Level.SUB_ACCOUNT,
        tenant: Optional[str] = None,
    ) -> None: ...

Models

  • Destination(name: str, type: str, url?: str, proxy_type?: str, authentication?: str, description?: str, properties?: dict[str, str], auth_tokens?: list[AuthToken], certificates?: list[Certificate])
    • auth_tokens and certificates are populated by the v2 consumption API
  • ConsumptionOptions - Options for v2 destination consumption, controls HTTP headers sent to the Destination Service:
    • fragment_name?: str - Fragment to merge into the destination (X-fragment-name)
    • fragment_level?: ConsumptionLevel - Level hint for the fragment lookup; appended to fragment_name as @level (e.g., "my-frag@provider_subaccount"). Only effective when fragment_name is also provided.
    • fragment_optional?: bool - If True, a missing fragment does not cause an error (X-fragment-optional)
    • tenant?: str - Tenant subdomain for token retrieval (X-tenant)
    • user_token?: str - User JWT for OAuth2UserTokenExchange / OAuth2JWTBearer / OAuth2SAMLBearerAssertion (X-user-token)
    • subject_token?: str - Subject token for OAuth2TokenExchange (X-subject-token)
    • subject_token_type?: str - Format of the subject token (X-subject-token-type), e.g. "urn:ietf:params:oauth:token-type:access_token"
    • actor_token?: str - Actor token for OAuth2TokenExchange (X-actor-token)
    • actor_token_type?: str - Format of the actor token (X-actor-token-type)
    • saml_assertion?: str - Client-provided SAML assertion for OAuth2SAMLBearerAssertion with SAMLAssertionProvider=ClientProvided (X-samlAssertion)
    • refresh_token?: str - Refresh token for OAuth2RefreshToken destinations (X-refresh-token)
    • code?: str - Authorization code for OAuth2AuthorizationCode destinations (X-code)
    • redirect_uri?: str - Redirect URI for OAuth2AuthorizationCode destinations (X-redirect-uri)
    • code_verifier?: str - PKCE code verifier for OAuth2AuthorizationCode destinations (X-code-verifier)
    • chain_name?: str - Name of a predefined destination chain (X-chain-name)
    • chain_vars?: dict[str, str] - Variables for the destination chain; each entry is sent as X-chain-var-<key>
  • AuthToken(type: str, value: str, http_header: dict, expires_in?: str, error?: str, scope?: str, refresh_token?: str) - Authentication token from v2 API
  • Fragment(name: str, properties: dict[str, str])
  • Certificate(name: str, content: str, type: str)
  • DestinationConfig(url, token_url, client_id, client_secret, identityzone)
  • TransparentProxy(proxy_name: str, namespace: str) - Configuration for transparent proxy routing
  • TransparentProxyDestination(name: str, url: str, headers: dict[str, str]) - Destination configured for transparent proxy access
  • Label(key: str, values: List[str]) - Key-value metadata tag for filtering and organizing resources
  • PatchLabels(action: str, labels: List[Label]) - Incremental label update; action is "ADD" (upsert) or "DELETE" (remove)
  • ListOptions(filter_names?: List[str], filter_labels?: List[Label], page?: int, page_size?: int, page_count?: bool, entity_count?: bool) - Filtering and pagination for list operations; filter_labels uses an OData Label HAS expression
  • PagedResult[T](items: list[T], pagination?: PaginationInfo) - Contains results and optional pagination metadata
  • PaginationInfo(next_cursor?: str, total_count?: int) - Pagination metadata from response headers

Notes:

  • Unknown string-valued destination fields are captured into Destination.properties preserving their original key casing and are included when serializing via Destination.to_dict. Non-string unknown fields are ignored.
  • Fragment properties are stored as string key-value pairs in Fragment.properties.
  • 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

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 headerssap-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 = secretapiKey: 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:

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:

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.

Configuration

There are three ways to configure transparent proxy support:

1. Client-Level Default (Recommended)

Enable proxy by default for all destination lookups when creating the client:

from sap_cloud_sdk.destination import create_client

# Default: use_default_proxy=False
# Turning it to true will use TransparentProxy from APPFND_CONHOS_TRANSP_PROXY environment variable
client = create_client(instance="default", use_default_proxy=True)

# All get operations will use the proxy by default
dest = client.get_instance_destination("my-destination")
# Returns TransparentProxyDestination

The environment variable APPFND_CONHOS_TRANSP_PROXY should be set with the format {proxy_name}.{namespace}:

export APPFND_CONHOS_TRANSP_PROXY="connectivity-proxy.my-namespace"

This setting might be automatically configured depending on the runtime

2. Explicit Proxy Configuration

You can set or update the proxy configuration after client creation using the set_proxy() method:

from sap_cloud_sdk.destination import create_client, TransparentProxy

# Create client first
client = create_client(instance="default")

# Set custom proxy configuration
transparent_proxy = TransparentProxy(
    proxy_name="my-destination", namespace="my-namespace"
)
client.set_proxy(transparent_proxy)

3. Per-Request Override

Override the client's default proxy setting for individual requests:

# Client created with use_default_proxy=True (uses proxy by default)
client = create_client(instance="default", use_default_proxy=True)

# Override to NOT use proxy for this specific request
dest = client.get_instance_destination("my-destination", proxy_enabled=False)
# Returns regular Destination

# Or explicitly enable proxy (even if client default is False)
client2 = create_client(instance="default", use_default_proxy=False)
dest2 = client2.get_instance_destination("my-destination", proxy_enabled=True)
# Returns TransparentProxyDestination

Usage Examples

from sap_cloud_sdk.destination import create_client, TransparentProxy, AccessStrategy

# Example 1: Using environment variable with default proxy
client = create_client(instance="default", use_default_proxy=True)
dest = client.get_instance_destination("my-destination")
# Uses proxy from APPFND_CONHOS_TRANSP_PROXY

# Example 2: Explicit proxy configuration with set_proxy()
client = create_client(instance="default")
transparent_proxy = TransparentProxy(proxy_name="my-proxy", namespace="my-namespace")
client.set_proxy(transparent_proxy)
dest = client.get_instance_destination("my-destination", proxy_enabled=True)

# Example 3: Update proxy after creation
client = create_client(instance="default")
transparent_proxy = TransparentProxy(
    proxy_name="my-destination", namespace="my-namespace"
)
client.set_proxy(transparent_proxy)
dest = client.get_instance_destination("my-destination", proxy_enabled=True)

# Example 4: Subaccount destination with proxy
client = create_client(instance="default", use_default_proxy=True)
dest = client.get_subaccount_destination(
    name="my-destination", access_strategy=AccessStrategy.PROVIDER_ONLY
)
# Uses proxy by default (client's use_default_proxy=True)

# Example 5: Override client default per request
client = create_client(instance="default", use_default_proxy=True)
regular_dest = client.get_instance_destination("my-destination", proxy_enabled=False)
# Returns regular Destination (overrides client default)

# Example 6: V2 API (get_destination) with proxy support
client = create_client(instance="default", use_default_proxy=True)
dest = client.get_destination("my-api", proxy_enabled=True)
# Returns TransparentProxyDestination

# Example 7: V2 API with ConsumptionOptions and proxy disabled
client = create_client(instance="default", use_default_proxy=True)
options = ConsumptionOptions(fragment_name="production", tenant="tenant-1")
dest = client.get_destination("my-api", options=options, proxy_enabled=False)
# Returns regular Destination with merged fragment and tenant context

# Example 8: V2 API with level parameter for optimized lookup
client = create_client(instance="default")
# Search only at provider subaccount level
dest = client.get_destination("my-api", level=ConsumptionLevel.PROVIDER_SUBACCOUNT)
# Search only at subscriber subaccount level
dest = client.get_destination("my-api", level=ConsumptionLevel.SUBACCOUNT)
# Search only at service instance level
dest = client.get_destination("my-api", level=ConsumptionLevel.INSTANCE)
# No level specified - API resolves automatically
dest = client.get_destination("my-api")

# Example 9: Combine level with options
options = ConsumptionOptions(fragment_name="production", tenant="tenant-1")
dest = client.get_destination(
    "my-api", level=ConsumptionLevel.SUBACCOUNT, options=options
)

# Example 9b: Fragment with level hint
options = ConsumptionOptions(
    fragment_name="my-fragment",
    fragment_level=ConsumptionLevel.PROVIDER_SUBACCOUNT,
)
dest = client.get_destination("my-api", options=options)

# Example 10: Optional fragment (no error if fragment does not exist)
options = ConsumptionOptions(fragment_name="maybe-exists", fragment_optional=True)
dest = client.get_destination("my-api", options=options)

# Example 11: User token exchange (OAuth2UserTokenExchange / OAuth2JWTBearer)
options = ConsumptionOptions(user_token="<encoded-jwt>", tenant="tenant-1")
dest = client.get_destination("my-api", options=options)

# Example 12: OAuth2TokenExchange with subject and actor tokens
options = ConsumptionOptions(
    subject_token="<subject-token>",
    subject_token_type="urn:ietf:params:oauth:token-type:access_token",
    actor_token="<actor-token>",
    actor_token_type="urn:ietf:params:oauth:token-type:access_token",
)
dest = client.get_destination("my-api", options=options)

# Example 13: Client-provided SAML assertion (SAMLAssertionProvider=ClientProvided)
options = ConsumptionOptions(saml_assertion="<base64-encoded-saml>")
dest = client.get_destination("my-api", options=options)

# Example 14: OAuth2RefreshToken
options = ConsumptionOptions(refresh_token="<refresh-token>")
dest = client.get_destination("my-api", options=options)

# Example 15: OAuth2AuthorizationCode with PKCE
options = ConsumptionOptions(
    code="<authorization-code>",
    redirect_uri="https://myapp/callback",
    code_verifier="<pkce-code-verifier>",
)
dest = client.get_destination("my-api", options=options)

# Example 16: Destination chain with chain variables
options = ConsumptionOptions(
    chain_name="my-predefined-chain",
    chain_vars={"subject_token": "<token>", "subject_token_type": "access_token"},
)
dest = client.get_destination("my-api", options=options)

Return Type

When proxy_enabled=True (either as client default or per-request override), the methods return a TransparentProxyDestination instead of a regular Destination:

# TransparentProxyDestination has these properties:
# - name: str - The destination name
# - url: str - The proxy URL (e.g., "http://connectivity-proxy.my-namespace")
# - headers: dict[str, str] - Required headers including "X-destination-name"

client = create_client(instance="default", use_default_proxy=True)
proxy_dest = client.get_instance_destination("my-destination")
print(proxy_dest.name)  # "my-destination"
print(proxy_dest.url)  # "http://connectivity-proxy.my-namespace"
print(proxy_dest.headers)  # {"X-destination-name": "my-destination"}

Setting Custom Headers

The TransparentProxyDestination class provides a set_header() method to add or update headers required by the transparent proxy. Use the TransparentProxyHeader enum to ensure type-safe header names:

from sap_cloud_sdk.destination import create_client, TransparentProxyHeader

# Get a transparent proxy destination
client = create_client(instance="default", use_default_proxy=True)
proxy_dest = client.get_instance_destination("my-destination")

# Set additional headers using the enum
proxy_dest.set_header(TransparentProxyHeader.AUTHORIZATION, "Bearer token123")

# Access the updated headers
print(proxy_dest.headers)
# {
#   "X-destination-name": "my-destination",
#   "Authorization": "Bearer token123"
# }

Available Headers (TransparentProxyHeader enum):

  • X_DESTINATION_NAME - Header for specifying the destination name (automatically set by from_proxy())
  • AUTHORIZATION - Header for authorization (e.g., "Bearer token", "Basic base64credentials")
  • X_FRAGMENT_NAME - Header for specifying the fragment name
  • X_TENANT_SUBDOMAIN - Header for tenant subdomain
  • X_TENANT_ID - Header for tenant ID
  • X_FRAGMENT_OPTIONAL - Header for optional fragment flag
  • X_DESTINATION_LEVEL - Header for destination level
  • X_FRAGMENT_LEVEL - Header for fragment level
  • X_TOKEN_SERVICE_TENANT - Header for token service tenant
  • X_CLIENT_ASSERTION - Header for client assertion
  • X_CLIENT_ASSERTION_TYPE - Header for client assertion type
  • X_CLIENT_ASSERTION_DESTINATION_NAME - Header for client assertion destination name
  • X_SUBJECT_TOKEN_TYPE - Header for subject token type
  • X_ACTOR_TOKEN - Header for actor token
  • X_ACTOR_TOKEN_TYPE - Header for actor token type
  • X_REDIRECT_URI - Header for redirect URI
  • X_CODE_VERIFIER - Header for code verifier
  • X_CHAIN_NAME - Header for chain name
  • X_CHAIN_VAR_SUBJECT_TOKEN - Header for chain variable subject token
  • X_CHAIN_VAR_SUBJECT_TOKEN_TYPE - Header for chain variable subject token type
  • X_CHAIN_VAR_SAML_PROVIDER_DESTINATION_NAME - Header for chain variable SAML provider destination name

The set_header() method accepts:

  • header: A TransparentProxyHeader enum value
  • value: The string value for the header

This ensures only valid headers are used with transparent proxy destinations.

Important Notes

  • If use_default_proxy=True but no proxy configuration is available in the environment variable, load_transparent_proxy() returns None and proxy functionality is disabled
  • The actual destination configuration is retrieved by the proxy service, not by the SDK
  • When proxy_enabled is not specified in get methods, the client's default setting (from use_default_proxy) is used
  • Proxy support is available for all three get methods:
    • get_instance_destination() - V1 API for instance-level destinations
    • get_subaccount_destination() - V1 API for subaccount-level destinations with access strategies
    • get_destination() - V2 API for runtime consumption with automatic token retrieval

Label Management

Labels are key-value metadata tags that can be attached to destinations, fragments, and certificates. They enable filtering resources by label values using ListOptions.filter_labels.

Models

  • Label(key: str, values: List[str]) — a label with one or more string values (e.g., Label(key="env", values=["prod", "eu"]))
  • PatchLabels(action: str, labels: List[Label]) — incremental update; action="ADD" upserts label values, action="DELETE" removes them

Operations

Each client (DestinationClient, FragmentClient, CertificateClient) exposes three label methods:

Method HTTP Description
get_*_labels(name, level, tenant) GET Returns current labels
update_*_labels(name, labels, level, tenant) PUT Replaces all labels atomically
patch_*_labels(name, patch, level, tenant) PATCH Adds or removes specific labels

All three accept an optional tenant parameter (like the create/update/delete methods) to scope the request to a subscriber context.

Examples

from sap_cloud_sdk.destination import (
    create_client,
    create_fragment_client,
    create_certificate_client,
    Label,
    PatchLabels,
    ListOptions,
    Level,
)

client = create_client(instance="default")
fragment_client = create_fragment_client(instance="default")
certificate_client = create_certificate_client(instance="default")

# Get labels for a destination (provider context)
labels = client.get_destination_labels("my-dest", level=Level.SUB_ACCOUNT)

# Get labels in subscriber context
labels = client.get_destination_labels(
    "my-dest", level=Level.SUB_ACCOUNT, tenant="tenant-subdomain"
)

# Replace all labels (PUT)
new_labels = [Label(key="env", values=["prod"]), Label(key="team", values=["platform"])]
client.update_destination_labels(
    "my-dest", new_labels, level=Level.SUB_ACCOUNT, tenant="tenant-subdomain"
)

# Add labels incrementally (PATCH ADD — upserts existing keys)
client.patch_destination_labels(
    "my-dest",
    PatchLabels(action="ADD", labels=[Label(key="region", values=["eu"])]),
    level=Level.SUB_ACCOUNT,
    tenant="tenant-subdomain",
)

# Remove specific labels (PATCH DELETE)
client.patch_destination_labels(
    "my-dest",
    PatchLabels(action="DELETE", labels=[Label(key="region", values=["eu"])]),
    level=Level.SUB_ACCOUNT,
)

# Fragment and certificate label operations follow the same pattern
fragment_client.update_fragment_labels(
    "my-fragment", new_labels, level=Level.SUB_ACCOUNT, tenant="tenant-subdomain"
)
certificate_client.patch_certificate_labels(
    "my-cert.pem",
    PatchLabels(action="ADD", labels=[Label(key="env", values=["staging"])]),
    level=Level.SUB_ACCOUNT,
)

# Filter list results by label
from sap_cloud_sdk.destination import ListOptions

filter_opts = ListOptions(filter_labels=[Label(key="env", values=["prod"])])
result = client.list_subaccount_destinations(filter=filter_opts)
fragments = fragment_client.list_instance_fragments(filter=filter_opts)

Local Development Mode

When a mocks/<resource>.json file is present at the repository root, the factory functions automatically return a local in-memory client backed by that file instead of connecting to the SAP BTP Destination Service. No credentials or network access are required.

Factory Mock file
create_client() mocks/destination.json
create_fragment_client() mocks/fragments.json
create_certificate_client() mocks/certificates.json

WARNING: Local mode is for local development only. Local clients perform no authentication and store data in plain JSON files on disk. Never use local mode in a deployed or production environment. A warning is logged at WARNING level every time a local client is returned by a factory.

Recommended: add mock files to .gitignore

Mock files may contain sensitive data (URLs, credentials, certificates). Add them to .gitignore to prevent accidental commits:

mocks/destination.json
mocks/fragments.json
mocks/certificates.json

Mock file format

mocks/destination.json

{
  "subaccount": [
    {
      "name": "my-destination",
      "type": "HTTP",
      "url": "https://example.com",
      "authentication": "NoAuthentication"
    },
    {
      "tenant": "my-tenant",
      "name": "subscriber-destination",
      "type": "HTTP",
      "url": "https://subscriber.example.com",
      "authentication": "NoAuthentication"
    }
  ],
  "instance": [
    {
      "name": "instance-destination",
      "type": "HTTP",
      "url": "https://instance.example.com"
    }
  ]
}

mocks/fragments.json

{
  "subaccount": [
    {
      "FragmentName": "my-fragment",
      "URL": "https://example.com",
      "Authentication": "NoAuthentication"
    }
  ],
  "instance": []
}

mocks/certificates.json

{
  "subaccount": [
    {
      "Name": "my-cert.pem",
      "Content": "LS0tLS1CRUdJTi...",
      "Type": "PEM"
    }
  ],
  "instance": []
}

Entries with a "tenant" field are treated as subscriber-specific. Entries without "tenant" are provider entries.

Multi-tenancy

  • Supported: Yes
  • Authentication: XSUAA
  • How to use: Pass access_strategy and tenant to the methods. The AccessStrategy enum controls whether the SDK fetches a subscriber-scoped or provider-scoped token by replacing the identityzone subdomain in the token URL.
  • Further reading:

Error Handling

  • DestinationNotFoundError: mapped from HTTP 404 where applicable
  • DestinationOperationError: general operation failures
  • HttpError: HTTP-related or local store read/write errors with status_code and response_text when applicable

Configuration

Service Binding

  • Mount path: $SERVICE_BINDING_ROOT/destination/{instance}/ (defaults to /etc/secrets/appfnd/destination/{instance}/)
  • Required Keys*: clientid, clientsecret, url (auth base), uri (service base), identityzone
  • Env var fallback: CLOUD_SDK_CFG_DESTINATION_{INSTANCE}_{FIELD} (uppercased, hyphens in instance replaced with _)

Note: SERVICE_BINDING_ROOT defaults to /etc/secrets/appfnd when not set. See the Secret Resolver guide for details.

Mounted Secrets (Kubernetes)

$SERVICE_BINDING_ROOT/destination/{instance}/
├── clientid
├── clientsecret
├── url
├── uri
└── identityzone

Environment Variables

# Example for Destination with instance name "default"
export CLOUD_SDK_CFG_DESTINATION_DEFAULT_CLIENTID="your-client-id"
export CLOUD_SDK_CFG_DESTINATION_DEFAULT_CLIENTSECRET="your-client-secret"
export CLOUD_SDK_CFG_DESTINATION_DEFAULT_URL="https://subdomain.authentication.region.hana.ondemand.com"
export CLOUD_SDK_CFG_DESTINATION_DEFAULT_URI="https://destination.cf.region.hana.ondemand.com"
export CLOUD_SDK_CFG_DESTINATION_DEFAULT_IDENTITYZONE="subdomain"

Transparent Proxy

  • Environment variable: APPFND_CONHOS_TRANSP_PROXY
  • Format: {proxy_name}.{namespace} (e.g., connectivity-proxy.my-namespace)
  • The proxy configuration is loaded and validated when the client is created
  • Proxy reachability is tested via HTTP HEAD request to http://{proxy_name}.{namespace}

Tokens and Access Strategy

The OAuth2 token URL is derived from service binding (DestinationConfig.token_url). For subscriber context, when a tenant is provided, the token provider constructs the subscriber token URL by replacing the identityzone segment with the tenant sub-domain.

Notes

  • Current implementation omits explicit HTTP retries/timeouts for simplicity.
  • The v2 consumption API (get_destination) is supported for runtime scenarios requiring automatic token retrieval.

Utilities

get_service_instance_id()

Reads the destination service instance ID from the mounted secret binding. This is useful when you need the instance ID for programmatic use (e.g., to pass it to another service or for logging).

from sap_cloud_sdk.destination import get_service_instance_id

instance_id = get_service_instance_id()

The function reads the instanceid field from the secret resolved via the standard mount/env fallback:

  • Mount path: /etc/secrets/appfnd/destination/default/instanceid
  • Env var fallback: CLOUD_SDK_CFG_DESTINATION_DEFAULT_INSTANCEID

Raises DestinationOperationError if the secret cannot be resolved (e.g., running locally without a binding).