Consuming on-prem OData service though Destination in python script and deploy it to CF in BTP.

[s_satyamsetti]

Hello SAP Community,

I have developed a python script in Business Application Studio in SAP BTP and in that I am trying to consume an on premise OData service through destination.
If am giving my backend Host & port URL I am able to consume the data directly, but it would not work if I deploy it as my service is not exposed outside.
Can you please help me how to call the OData service through destination in Python Script similar to how we call in SAP UI5 applications.

Thanks & Regards,

[Bibhu]

A sample implementation of Consuming the Destination Service "Destination Service REST API"

1. Set up a Destination Service Instance for your Subaccount

cf create-service destination lite <INSTANCE_NAME>​

2. Get Credentials

#After creating the Destination service instance, create a service key for it to retrieve the necessary credentials:
cf create-service-key <destination-service-instance-name> <service-key-name>

#Retrieve the service key credentials
cf service-key <destination-service-instance-name> <service-key-name>​

3. Get an AccessToken

TOKEN_PATH = "/oauth/token"

def get_access_token():
    """Fetch OAuth token from the Destination service using credentials from VCAP_SERVICES."""
    client_id = os.getenv('CLIENT_ID')
    client_secret = os.getenv('CLIENT_SECRET')
    token_url = os.getenv('TOKEN_URL') + TOKEN_PATH  # Append /oauth/token for the token endpoint

    # OAuth token request
    data = {
        'grant_type': 'client_credentials',
        'client_id': client_id,
        'client_secret': client_secret
    }
    headers = {'Content-Type': 'application/x-www-form-urlencoded'}

    response = requests.post(token_url, data=data, headers=headers)
    response.raise_for_status()
    return response.json().get('access_token')​

4. Call the Destination Service REST API

def get_destination_details(token):
    """Fetch destination details from the binding."""
    # Get the destination name from the .env file
    destination_name = os.getenv('DESTINATION_NAME')
    
    if not destination_name:
        raise Exception("DESTINATION_NAME environment variable is not set")

    # Get the VCAP_SERVICES environment variable
    vcap_services = os.getenv('VCAP_SERVICES')
    
    if not vcap_services:
        raise Exception("VCAP_SERVICES environment variable is not set")
    
    # Parse the JSON binding to retrieve Destination service details
    services = json.loads(vcap_services)
    
    # Access the destination service configuration
    destination_service = services.get('destination')
    
    if not destination_service:
        raise Exception("No destination service found in VCAP_SERVICES")
    
    # Assuming we want the first destination service instance
    destination_instance = destination_service[0]['credentials']
    uri = destination_instance.get('uri')
    
    # Call the Destination service REST API to fetch destinations
    destinations = call_destination_service_api(uri, token)

    if not destinations:
        raise Exception("No destinations found")

    # Filter the destinations based on the destination name from the .env file
    filtered_destination = next((dest for dest in destinations if dest['Name'] == destination_name), None)

    if not filtered_destination:
        raise Exception(f"Destination with name '{destination_name}' not found")

    return filtered_destination  # Return the filtered destination instance

def call_destination_service_api(uri, token):
    """Call the Destination service REST API to get the list of destinations."""
    response = requests.get(f"{uri}{DESTINATION_PATH}", 
                            headers={'Authorization': f'Bearer {token}'})
    response.raise_for_status()
    return response.json()

5. Call the OData service (works ProxyType “Internet”) similarly you can implement On-Premise

def call_odata_service(odata_url, token, destination):
    """Call the OData service via the destination."""
    auth_type = destination.get('Authentication')
    
    headers = {'Authorization': f'Bearer {token}'}

    if auth_type == "BasicAuthentication":
        username = destination.get('User')
        password = destination.get('Password')
        response = requests.get(odata_url, auth=(username, password), headers=headers)
    else:
        response = requests.get(odata_url, headers=headers)
    
    response.raise_for_status()
    return response

6. Call the OData service (works ProxyType “On-Premise”)

WIP

There are countless possibilities for improvement, but this represents the overall implementation. Please find the sample git link, which I initiated as a small project a while ago and finally pushed today.

@gregorw Thank you so much for sharing valuable information on how to use the Consuming the Connectivity Service. Your insight is truly appreciated and will be incredibly helpful.

[gregorw]

I would suggest you read the documentation Consuming the Connectivity Service.