Loading app/app_template.py +20 −12 Original line number Diff line number Diff line Loading @@ -3,7 +3,8 @@ import json import pickle import requests from flask import Flask, jsonify, request # import OpenCAPIF SDK ### Import OpenCAPIF SDK package # import opencapif_sdk app = Flask(__name__) OBJ_FILE = "discoverer.pkl" Loading @@ -16,18 +17,21 @@ user_name = os.getenv("USER_NAME", "test") def onboard(): response_message = "Onboard" ### Create the connection with CAPIF and set the invoker's config file # invoker = opencapif_sdk.capif_invoker_connector(config_file="invoker_sdk_config.json") ### Onboard the invoker # invoker.onboard_invoker() ### Uncomment to print the Invoker ID ### Uncomment to get the Invoker ID # with open("./invoker_info/{0}/capif_api_security_context_details-{0}.json".format(user_name), "rb") as f: # d = json.load(f) # invoker_id = d['api_invoker_id'] ### Uncomment, fill the ... with the object and store invoker object ### Uncomment to store invoker object # with open(OBJ_FILE_INV, 'wb') as f: # pickle.dump(invoker, f) # response_message = "Invoker onboarded with ID: {}".format(d['api_invoker_id']) # response_message = "Invoker onboarded with ID: {}".format(d['invoker_id']) return response_message Loading @@ -36,13 +40,14 @@ def onboard(): def discover(): response_message = "Discover" ### Create the connection with CAPIF as an onboarded invoker # service_discoverer = ... # service_discoverer = opencapif_sdk.service_discoverer(config_file="invoker_sdk_config.json") ### Discover APIs according to the filters # service_discoverer.discover() ### Uncomment, fill the ... with the object and store discoverer object ### Uncomment to store discoverer object # with open(OBJ_FILE, 'wb') as f: # pickle.dump( ... , f) # pickle.dump( service_discoverer , f) ### Uncomment to read and return the discovered API # with open("./invoker_info/{0}/capif_api_security_context_details-{0}.json".format(user_name), "rb") as f: Loading @@ -68,16 +73,18 @@ def access_service(): # service_discoverer = pickle.load(f) ### Fetch the necessary information of the targeted API and add to a url parameter # service_path = ... # api_details = ... # service_interface = # url = ... # service_path = service_discoverer.retrieve_specific_resource_name(api_name, resource_name).split("/")[-1] # api_details = service_discoverer.retrieve_api_description_by_name(api_name) # service_interface = api_details["aefProfiles"][0]['interfaceDescriptions'][0] # url = "http://{}:{}/{}".format(service_interface['ipv4Addr'], service_interface['port'], service_path) ### Get API token for the discovered API # service_discoverer.get_tokens() # jwt_token = service_discoverer.token ### Uncomment to add token to a header # headers = { # 'Authorization': 'Bearer {}'.format(...) # 'Authorization': 'Bearer {}'.format(jwt_token) # } ### Uncomment to make the request and return response text Loading @@ -97,6 +104,7 @@ def offboard(): # invoker = pickle.load(f) ### Offboard invoker # invoker.offboard_invoker() ### Uncomment to check if invoker was successfully offboarded # if not os.path.exists("./invoker_info/{}".format(user_name)): Loading Loading
app/app_template.py +20 −12 Original line number Diff line number Diff line Loading @@ -3,7 +3,8 @@ import json import pickle import requests from flask import Flask, jsonify, request # import OpenCAPIF SDK ### Import OpenCAPIF SDK package # import opencapif_sdk app = Flask(__name__) OBJ_FILE = "discoverer.pkl" Loading @@ -16,18 +17,21 @@ user_name = os.getenv("USER_NAME", "test") def onboard(): response_message = "Onboard" ### Create the connection with CAPIF and set the invoker's config file # invoker = opencapif_sdk.capif_invoker_connector(config_file="invoker_sdk_config.json") ### Onboard the invoker # invoker.onboard_invoker() ### Uncomment to print the Invoker ID ### Uncomment to get the Invoker ID # with open("./invoker_info/{0}/capif_api_security_context_details-{0}.json".format(user_name), "rb") as f: # d = json.load(f) # invoker_id = d['api_invoker_id'] ### Uncomment, fill the ... with the object and store invoker object ### Uncomment to store invoker object # with open(OBJ_FILE_INV, 'wb') as f: # pickle.dump(invoker, f) # response_message = "Invoker onboarded with ID: {}".format(d['api_invoker_id']) # response_message = "Invoker onboarded with ID: {}".format(d['invoker_id']) return response_message Loading @@ -36,13 +40,14 @@ def onboard(): def discover(): response_message = "Discover" ### Create the connection with CAPIF as an onboarded invoker # service_discoverer = ... # service_discoverer = opencapif_sdk.service_discoverer(config_file="invoker_sdk_config.json") ### Discover APIs according to the filters # service_discoverer.discover() ### Uncomment, fill the ... with the object and store discoverer object ### Uncomment to store discoverer object # with open(OBJ_FILE, 'wb') as f: # pickle.dump( ... , f) # pickle.dump( service_discoverer , f) ### Uncomment to read and return the discovered API # with open("./invoker_info/{0}/capif_api_security_context_details-{0}.json".format(user_name), "rb") as f: Loading @@ -68,16 +73,18 @@ def access_service(): # service_discoverer = pickle.load(f) ### Fetch the necessary information of the targeted API and add to a url parameter # service_path = ... # api_details = ... # service_interface = # url = ... # service_path = service_discoverer.retrieve_specific_resource_name(api_name, resource_name).split("/")[-1] # api_details = service_discoverer.retrieve_api_description_by_name(api_name) # service_interface = api_details["aefProfiles"][0]['interfaceDescriptions'][0] # url = "http://{}:{}/{}".format(service_interface['ipv4Addr'], service_interface['port'], service_path) ### Get API token for the discovered API # service_discoverer.get_tokens() # jwt_token = service_discoverer.token ### Uncomment to add token to a header # headers = { # 'Authorization': 'Bearer {}'.format(...) # 'Authorization': 'Bearer {}'.format(jwt_token) # } ### Uncomment to make the request and return response text Loading @@ -97,6 +104,7 @@ def offboard(): # invoker = pickle.load(f) ### Offboard invoker # invoker.offboard_invoker() ### Uncomment to check if invoker was successfully offboarded # if not os.path.exists("./invoker_info/{}".format(user_name)): Loading