python_app_filename.py:
It is the main python application file containing all the necessary codes.# importing modules
# ---------------------------------------------------------------------
# import all the necessary modules here that will be used in the program
from flask import Flask, request, jsonify
import os
# importing local module: BTPservices: containing generic codes
# ---------------------------------------------------------------------
import BTPservices
# application initialization
# ---------------------------------------------------------------------
app = Flask(__name__)
# fetching port details
# ---------------------------------------------------------------------
cf_port = os.getenv("PORT")
# fetching of records having OAuth2ClientCredentials authorization
# ---------------------------------------------------------------------
def get_records_01(lv_dest, lv_string):
lv_destination = BTPservices.getDestination(sDestinationService=service_name_001, sDestinationName=lv_dest)
lv_url = lv_destination['destinationConfiguration']['URL'] + lv_string
try:
lv_request = requests.get(lv_url, headers={'Accept': 'application/json'})
except requests.exceptions.HTTPError as err:
return jsonify(replies=err)
lv_request = lv_request.json()
# clearing variables
del lv_destination, lv_url
return lv_request.json()
# fetching of records having OAuth2Password/No authorizartion
# ---------------------------------------------------------------------
def get_records_02(lv_dest, lv_string, lv_flag):
lv_destination = BTPservices.getDestination(sDestinationService=service_name_001, sDestinationName=lv_dest)
# having authorization
if lv_flag == 'True':
lv_token = str(lv_destination['authTokens'][0]['http_header']['value'])
lv_headers = {'Authorization': lv_token}
lv_response = requests.get(lv_destination['destinationConfiguration']['URL'] + lv_string, headers=lv_headers)
else:
# having no authorization
lv_response = requests.get(lv_destination['destinationConfiguration']['URL'] + lv_string )
# clearing the value
del lv_destination, lv_flag, lv_token
# returns the data
return lv_response.json()
# data fetch using the destination
# ---------------------------------------------------------------------
@app.route('/fetch_data_01', methods=['GET'])
def fetch_data_01():
lv_request = get_records_01('destination_name', 'entity_name')
return lv_request
@app.route('/fetch_data_02', methods=['GET'])
def fetch_data_02():
lv_request = get_records_02('destination_name', 'entity_name', 'True')
return lv_request
# homepage
# ---------------------------------------------------------------------
@app.route('/', methods=['GET'])
def home():
lv_return = {
"result": [
{
'message': 'Welcome to the homepage!',
'author': 'Kallol Chakraborty',
'date': '11/10/2022 ',
'description': 'The is a demo application'
}
]
}
return jsonify(status='200', replies=lv_return)
# main
# ---------------------------------------------------------------------
if __name__ == '__main__':
# If the app is running locally
if cf_port is None:
# Use port 5000
app.run(host='0.0.0.0', port=5000, debug=True)
else:
# Else use cloud foundry default port
app.run(host='0.0.0.0', port=int(cf_port), debug=False)
manifest.yml:
The manifest.yml
file represents the configuration describing your application and how it will be deployed to Cloud Foundry.IMPORTANT: Make sure you don’t have another application with the name app_name
in your space. If you do, use a different name and adjust the whole tutorial according to it. Also bear in mind that your application’s technical name (in the route) must be unique in the whole Cloud Foundry landscape. The advice is that you use, for example, your subdomain name or part of your subaccount ID to construct the technical name. In this tutorial, we use: application-1234-abcd-007
. The path to access the application will be https://app-1234-abcd-007.cfapps.eu20.hana.ondemand.com
---
applications:
- name: app_name
routes:
- route: app-1234-abcd-007.cfapps.us10.hana.ondemand.com
path: ./
memory: 128M
disk_quota: 512M
buildpack: python_buildpack
command: python python_app_filename.py
services:
- service_name_001
service_name_001
is bound to the application now but it is not created yet. Please create the service: service_name_001
using the CF CLI command provided below.cf create-service destination lite service_name_001
ProcFile:
The Procfile
specifies the command line to execute the application at runtime.web: python my_app.py
runtime.txt:
Specify the Python runtime version that your application will run on.python-3.10.1
requirements.txt:
This application will be a web server utilizing the Flask web framework. To specify flask, requests, cfenv, etc. as an application dependencies, please check the below sample code.flask==2.0.1
requests
cfenv
BTP Service.py:
This is a very important file containing some generic codes & is responsible for fetching the details of the BTP destinations.# modules
# -----------------------------------------------------------------------------------------------------
from cfenv import AppEnv
import requests
def getDestination(sDestinationService, sDestinationName):
# Read the environment variables
# -----------------------------------------------------------------------------------------------------
env = AppEnv()
dest_service = env.get_service(name=sDestinationService)
sUaaCredentials = dest_service.credentials["clientid"] + ':' + dest_service.credentials["clientsecret"]
# Request a JWT token to access the destination service
# -----------------------------------------------------------------------------------------------------
headers = {'Authorization': 'Basic '+sUaaCredentials, 'content-type': 'application/x-www-form-urlencoded'}
form = [('client_id', dest_service.credentials["clientid"] ),('client_secret', dest_service.credentials["clientsecret"] ), ('grant_type', 'client_credentials')]
url = dest_service.credentials['url'] +"/oauth/token"
headers = {
'Content-Type': 'application/x-www-form-urlencoded'
}
response = requests.request("POST", url, headers=headers, data=form)
# Search your destination in the destination service
# -----------------------------------------------------------------------------------------------------
token = response.json()["access_token"]
headers= { 'Authorization': 'Bearer ' + token }
r = requests.get(dest_service.credentials["uri"] + '/destination-configuration/v1/destinations/'+sDestinationName, headers=headers)
print("DEST URI:",dest_service.credentials['uri'])
# Access the destination securely
# -----------------------------------------------------------------------------------------------------
destination = r.json()
return destination
You must be a registered user to add a comment. If you've already registered, sign in. Otherwise, register and sign in.
User | Count |
---|---|
11 | |
9 | |
6 | |
5 | |
4 | |
4 | |
4 | |
4 | |
3 | |
3 |