in orchestrator_client.py [0:0]
def get_function_key():
subscription_id = os.getenv('AZURE_SUBSCRIPTION_ID')
resource_group = os.getenv('AZURE_RESOURCE_GROUP_NAME')
function_app_name = os.getenv('AZURE_ORCHESTRATOR_FUNC_NAME')
token = get_managed_identity_token()
logging.info("[orchestrator_client] Obtaining function key.")
# URL to get all function keys, including the default one
requestUrl = f"https://management.azure.com/subscriptions/{subscription_id}/resourceGroups/{resource_group}/providers/Microsoft.Web/sites/{function_app_name}/functions/orchestrator_streaming/listKeys?api-version=2022-03-01"
requestHeaders = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
response = requests.post(requestUrl, headers=requestHeaders)
# Check for HTTP errors
if response.status_code >= 400:
logging.error(f"[orchestrator_client] Failed to obtain function key. HTTP status code: {response.status_code}. Error details: {response.text}")
function_key = None
else:
try:
response_json = response.json()
function_key = response_json['default']
except KeyError as e:
function_key = None
logging.error(f"[orchestrator_client] Error when getting function key. Details: {str(e)}.")
return function_key