70 строки
2.9 KiB
Python
70 строки
2.9 KiB
Python
import logging
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Dict
|
|
|
|
from azureml.core import Workspace
|
|
from injector import singleton, Binder
|
|
|
|
from azure_config import AzureConfig
|
|
from configuration_constants import (API_AUTH_SECRET_ENVIRONMENT_VARIABLE, CLUSTER, WORKSPACE_NAME,
|
|
EXPERIMENT_NAME, RESOURCE_GROUP, SUBSCRIPTION_ID,
|
|
APPLICATION_ID, TENANT_ID, IMAGE_DATA_FOLDER, DATASTORE_NAME,
|
|
AZUREML_SERVICE_PRINCIPAL_SECRET_ENVIRONMENT_VARIABLE)
|
|
|
|
PROJECT_SECRETS_FILE = Path(__file__).resolve().parent / Path("set_environment.sh")
|
|
|
|
|
|
def read_secret_from_file(secret_name: str) -> str:
|
|
"""
|
|
Reads a bash file with exports and returns the variables and values as dict
|
|
:return: A dictionary with secrets, or None if the file does not exist.
|
|
"""
|
|
try:
|
|
secrets_file = PROJECT_SECRETS_FILE
|
|
d: Dict[str, str] = {}
|
|
for line in secrets_file.read_text().splitlines():
|
|
if line.startswith("#"): continue
|
|
parts = line.replace("export", "").strip().split("=", 1)
|
|
key = parts[0].strip().upper()
|
|
d[key] = parts[1].strip()
|
|
return d[secret_name]
|
|
except Exception as ex:
|
|
logging.error(f"Missing configuration '{secret_name}'")
|
|
raise ex
|
|
|
|
|
|
def get_environment_variable(environment_variable_name: str) -> str:
|
|
value = os.environ.get(environment_variable_name, None)
|
|
if value is None:
|
|
value = read_secret_from_file(environment_variable_name)
|
|
if value is None:
|
|
raise ValueError(environment_variable_name)
|
|
return value
|
|
|
|
|
|
# AUTHENTICATION SECRET
|
|
API_AUTH_SECRET = get_environment_variable(API_AUTH_SECRET_ENVIRONMENT_VARIABLE)
|
|
API_AUTH_SECRET_HEADER_NAME = "API_AUTH_SECRET"
|
|
|
|
|
|
def configure(binder: Binder) -> None:
|
|
azure_config = get_azure_config()
|
|
workspace = azure_config.get_workspace()
|
|
binder.bind(Workspace, to=workspace, scope=singleton)
|
|
binder.bind(AzureConfig, to=azure_config, scope=singleton)
|
|
|
|
|
|
def get_azure_config() -> AzureConfig:
|
|
return AzureConfig(cluster=get_environment_variable(CLUSTER),
|
|
workspace_name=get_environment_variable(WORKSPACE_NAME),
|
|
experiment_name=get_environment_variable(EXPERIMENT_NAME),
|
|
resource_group=get_environment_variable(RESOURCE_GROUP),
|
|
subscription_id=get_environment_variable(SUBSCRIPTION_ID),
|
|
application_id=get_environment_variable(APPLICATION_ID),
|
|
service_principal_secret=get_environment_variable(
|
|
AZUREML_SERVICE_PRINCIPAL_SECRET_ENVIRONMENT_VARIABLE),
|
|
tenant_id=get_environment_variable(TENANT_ID),
|
|
datastore_name=get_environment_variable(DATASTORE_NAME),
|
|
image_data_folder=get_environment_variable(IMAGE_DATA_FOLDER))
|