时间戳

首页 > 解决方案 > 在 Python 中并发 set_secret 调用 Azure KeyVault API

python - 在 Python 中并发 set_secret 调用 Azure KeyVault API

问题描述

我有一个 python 脚本，它当前将秘密字符串写入 Azure KeyVault。如果列表很小，这很有效，但我有 4000+ KVs 要写入，每次管道运行需要 50 分钟。

我正在使用这样的库：

from azure.keyvault import KeyVaultClient, KeyVaultAuthentication, KeyVaultId
from azure.common.credentials import ServicePrincipalCredentials

def auth_callback(server, resource, scope):
...

client = KeyVaultClient(KeyVaultAuthentication(auth_callback))
data = sys.stdin.read()
if data:
    print('Writing outputs to keyvault '+kvURL)
    outputs = json.loads(data)
    n = 0
    for op in outputs.items():
        key = baseEnv + '-' + baseFunction + '-' + op[0]
        key = key.replace('_', '-')
        val = str(op[1]['value'])
        print('Storing: ' + key + ' = ' + val)
        try:
            secret_bundle = client.set_secret(kvURL, key, val)
        except:
            print('err in key'+key)
            exit(os.EX_SOFTWARE)
            n += 1
    print(str(n)+' secrets written.')
else:
    print('OK nothing to do.')

由于脚本只需要对 API 进行并发调用，因此我不需要明确需要多处理或线程并发，只是实例化客户端对象的异步方式。

文档对我帮助不大，而且我不知道任何 c#，所以我无法翻译我看到的任何异步方法。关于我应该如何处理这个欢迎的任何想法。特别是在实例化多个客户端对象时。

标签： pythonazureazure-keyvaultazure-sdk

解决方案

现在有用于在 Python 中使用 Key Vault 的新包替换azure-keyvault，每个包都支持使并发调用更直接的异步操作：

azure-identity也是应该与这些一起用于身份验证的包。

在秘密包 README中有异步创建秘密的文档；这是一个示例，说明如何将异步客户端用于您的用例（为了便于阅读，简化了一些位），利用asyncio 的gather方法：

import asyncio
from azure.identity.aio import DefaultAzureCredential
from azure.keyvault.secrets.aio import SecretClient

kvURL = "https://{vault-name}.vault.azure.net"
credential = DefaultAzureCredential()
client = SecretClient(kvURL, credential)

async def set_secrets():
    async with client:
        data = sys.stdin.read()
        if data:
            print("Writing outputs to keyvault " + kvURL)
            outputs = json.loads(data)
            n = 0
            coroutines = []
            for op in outputs.items():
                key = "secret" + str(n)
                val = "value"
                print('Storing: ' + key + ' = ' + val)
                try:
                    coroutines.append(client.set_secret(key, val))
                except:
                    print("err in key " + key)
                    exit(os.EX_SOFTWARE)
            return await asyncio.gather(*coroutines)
        else:
            print("OK nothing to do.")

loop = asyncio.get_event_loop()
results = loop.run_until_complete(set_secrets())
loop.close()
print(results)  # prints a list of secrets created

（我在 Python 中使用 Azure SDK）

推荐阅读