wehub-resource-sync
e768098d0e
Flake8 Lint / flake8 (push) Waiting to run
Spell check CI / Spell_Check (push) Waiting to run
tools_continuous_delivery / Private PyPI non-main branch release (push) Has been skipped
tools_continuous_delivery / Private PyPI main branch release (push) Failing after 2m42s
Publish Promptflow Doc / Build (push) Has been cancelled
Publish Promptflow Doc / Deploy (push) Has been cancelled
98 lines
2.8 KiB
Python
98 lines
2.8 KiB
Python
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from azure.keyvault.secrets import SecretClient
|
|
from azure.identity import ClientSecretCredential, DefaultAzureCredential
|
|
|
|
|
|
CONNECTION_FILE_NAME = "connections.json"
|
|
CONNECTION_TPL_FILE_PATH = Path(".") / "src/promptflow" / "dev-connections.json.example"
|
|
|
|
|
|
def get_secret_client(
|
|
tenant_id: str, client_id: str, client_secret: str
|
|
) -> SecretClient:
|
|
try:
|
|
if (tenant_id is None) or (client_id is None) or (client_secret is None):
|
|
credential = DefaultAzureCredential()
|
|
client = SecretClient(
|
|
vault_url="https://promptflowprod.vault.azure.net/",
|
|
credential=credential,
|
|
)
|
|
else:
|
|
credential = ClientSecretCredential(tenant_id, client_id, client_secret)
|
|
client = SecretClient(
|
|
vault_url="https://github-promptflow.vault.azure.net/",
|
|
credential=credential,
|
|
)
|
|
except Exception as e:
|
|
print(e)
|
|
|
|
return client
|
|
|
|
|
|
def get_secret(secret_name: str, client: SecretClient):
|
|
secret = client.get_secret(secret_name)
|
|
|
|
return secret.value
|
|
|
|
|
|
def list_secret_names(client: SecretClient) -> list:
|
|
secret_properties = client.list_properties_of_secrets()
|
|
|
|
return [secret.name for secret in secret_properties]
|
|
|
|
|
|
def fill_key_to_dict(template_dict, keys_dict):
|
|
if not isinstance(template_dict, dict):
|
|
return
|
|
for key, val in template_dict.items():
|
|
if isinstance(val, str) and val in keys_dict:
|
|
template_dict[key] = keys_dict[val]
|
|
continue
|
|
fill_key_to_dict(val, keys_dict)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument(
|
|
"--tenant_id", type=str, help="The tenant id of the service principal"
|
|
)
|
|
parser.add_argument(
|
|
"--client_id", type=str, help="The client id of the service principal"
|
|
)
|
|
parser.add_argument(
|
|
"--client_secret", type=str, help="The client secret of the service principal"
|
|
)
|
|
parser.add_argument(
|
|
"--target_folder", type=str, help="The target folder to save the generated file"
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
template_dict = json.loads(
|
|
open(CONNECTION_TPL_FILE_PATH.resolve().absolute(), "r").read()
|
|
)
|
|
file_path = (
|
|
(Path(".") / args.target_folder / CONNECTION_FILE_NAME)
|
|
.resolve()
|
|
.absolute()
|
|
.as_posix()
|
|
)
|
|
print(f"file_path: {file_path}")
|
|
|
|
client = get_secret_client(
|
|
tenant_id=args.tenant_id,
|
|
client_id=args.client_id,
|
|
client_secret=args.client_secret,
|
|
)
|
|
all_secret_names = list_secret_names(client)
|
|
data = {
|
|
secret_name: get_secret(secret_name, client) for secret_name in all_secret_names
|
|
}
|
|
|
|
fill_key_to_dict(template_dict, data)
|
|
|
|
with open(file_path, "w") as f:
|
|
json.dump(template_dict, f)
|