Azure App Configuration Python Provider客户端库

Azure App Configuration是一种帮助开发者简单、安全地集中管理应用程序配置的托管服务。此提供程序在azure-sdk-for-python之上添加了额外的功能。

使用此提供程序可以以托管的方式从Azure App Configuration存储加载配置集。

入门

获取凭据

使用以下Azure CLI片段从配置存储中获取连接字符串。

az appconfig credential list --name <config-store-name>

或者，从Azure门户获取连接字符串。

创建提供程序

您可以使用连接字符串创建客户端

from azure.appconfiguration.provider import load

config = load(connection_string="your-connection-string")

或者使用AAD

from azure.appconfiguration.provider import load

config = load(endpoint="your-endpoint", credential=DefaultAzureCredential())

默认情况下，这些提供程序将您的配置存储中所有具有（无标签）的配置加载到键/值字典中。

功能

目前，Azure App Configuration Provider 支持以下功能：

• 使用连接字符串或 Azure Active Directory 连接到 App Configuration 存储。
• 使用 SettingSelector 选择多组配置。
• 加载功能标志
• 动态刷新
• 地理复制支持
• 去除键名前缀。
• 解析密钥保管库引用，需要 AAD。
• 秘密解析器，本地解析密钥保管库引用，无需连接到密钥保管库。
• JSON 内容类型

未来功能

我们将要添加到 Python Provider 的功能列表。

• 配置占位符

示例

选择配置

您可以使用 SettingSelector 来细化或扩展从您的存储加载的配置。设置选择器提供了一种将键过滤器标签过滤器传递给提供者的方式。

from azure.appconfiguration.provider import load, SettingSelector
from azure.identity import DefaultAzureCredential

selects = {SettingSelector(key_filter="*", label_filter="\0"), SettingSelector(key_filter="*", label_filter="dev")}
config = load(endpoint=endpoint, credential=DefaultAzureCredential(), selects=selects)

在这个例子中，所有具有空标签和 dev 标签的配置都被加载。因为 dev 选择器列在最后，所以当出现重复时，来自 dev 的配置优先于具有 (No Label) 的配置。

动态刷新

提供者可以被配置为在设定的时间间隔内从存储刷新配置。这是通过向提供者提供一个 refresh_on 实现的，这是一个将监视变化的键（s）的列表，当它们发生变化时，将发生刷新。 refresh_interval 是刷新之间的时间间隔（以秒为单位）。on_refresh_success 是仅在检测到更改且没有错误发生时才会调用的回调。 on_refresh_error 是当刷新失败时才会调用的回调。

from azure.appconfiguration.provider import load, WatchKey
import os

connection_string = os.environ.get("APPCONFIGURATION_CONNECTION_STRING")

def my_callback_on_success():
    # Do something on success
    ...

def my_callback_on_fail(error):
    # Do something on fail
    ...

config = load(
    connection_string=connection_string,
    refresh_on=[WatchKey("Sentinel")],
    refresh_interval=60,
    on_refresh_success=my_callback_on_success,
    on_refresh_error=my_callback_on_fail,
    **kwargs,
)

在这个例子中，哨兵键将在每 60 秒之前不会更早地检查更改。为了检查更改，需要调用提供者的 refresh 方法。

config.refresh()

一旦提供者被刷新，就可以像通常一样访问配置。如果已经进行了更改，它将使用最新的值进行更新。如果自上次刷新检查以来没有经过 refresh_interval，提供者将不会检查更改。

有关更多信息，请参阅 MS Learn 上的 动态刷新。

修剪密钥

您可以通过向提供者提供一个要修剪的密钥前缀列表来修剪密钥。例如，如果您在配置存储中具有类似于 /application/message 的键（s），则可以从中修剪 /application/。

from azure.appconfiguration.provider import load
from azure.identity import DefaultAzureCredential

trim_prefixes={"/application/"}
config = load(endpoint=endpoint, credential=DefaultAzureCredential(), trim_prefixes=trim_prefixes)
print(config["message"])

解析密钥保管库引用

可以通过使用 AzureAppConfigurationKeyVaultOptions 将您的密钥保管库凭证提供给提供者来解析密钥保管库引用。

使用凭证

您可以通过提供凭证向 AzureAppConfigurationKeyVaultOptions，所有密钥保管库引用都将使用它进行解析。提供者将尝试使用提供的凭证连接到任何引用的密钥保管库。

from azure.appconfiguration.provider import load, AzureAppConfigurationKeyVaultOptions
from azure.identity import DefaultAzureCredential

key_vault_options = AzureAppConfigurationKeyVaultOptions(credential=DefaultAzureCredential())
config = load(endpoint=endpoint, credential=DefaultAzureCredential(), key_vault_options=key_vault_options)

使用客户端

您可以通过提供 AzureAppConfigurationKeyVaultOptions 中的 SecretClients 列表。

from azure.appconfiguration.provider import load, AzureAppConfigurationKeyVaultOptions
from azure.identity import DefaultAzureCredential

key_vault_options = AzureAppConfigurationKeyVaultOptions(
    client_configs={key_vault_uri: {'credential': credential}})
config = load(endpoint=endpoint, credential=DefaultAzureCredential(), key_vault_options=key_vault_options)

秘密解析器

如果没有提供凭证或客户端，可以使用秘密解析器。秘密解析器提供了一种返回您想要的任何值到密钥保管库引用的方式。

from azure.appconfiguration.provider import load, AzureAppConfigurationKeyVaultOptions
from azure.identity import DefaultAzureCredential

def secret_resolver(uri):
    return "From Secret Resolver"

key_vault_options = AzureAppConfigurationKeyVaultOptions(
    secret_resolver=secret_resolver)
config = load(endpoint=endpoint, credential=DefaultAzureCredential(), key_vault_options=key_vault_options)

地理复制

Azure App Configuration Provider 库将自动发现提供的配置存储的副本，并在出现任何问题时使用副本。更多信息请参阅 地理复制。

副本发现默认启用。如果您想禁用它，可以将 replica_discovery_enabled 设置为 False。

from azure.appconfiguration.provider import load
from azure.identity import DefaultAzureCredential

config = load(endpoint=endpoint, credential=DefaultAzureCredential(), replica_discovery_enabled=False)

加载功能标志

可以使用提供者从配置存储加载功能标志。功能标志作为存储在提供者 feature_management 下的字典加载，然后是 feature_flags。

config = load(endpoint=endpoint, credential=DefaultAzureCredential(), feature_flags_enabled=True)
alpha = config["feature_management"]["feature_flags"]["Alpha"]
print(alpha["enabled"])

默认情况下，当feature_flags_enabled设置为True时，所有未标记的特征标志将被加载。如果您想加载具有特定标记的特征标志，可以使用SettingSelector来过滤特征标志。

from azure.appconfiguration.provider import load, SettingSelector

config = load(endpoint=endpoint, credential=DefaultAzureCredential(), feature_flags_enabled=True, feature_flag_selectors=[SettingSelector(key_filter="*", label_filter="dev")])
alpha = config["feature_management"]["feature_flags"]["Alpha"]
print(alpha["enabled"])

要启用特征标志的刷新，您需要启用刷新。这将允许提供程序以刷新配置相同的方式刷新特征标志。与配置不同，所有加载的特征标志都会被监控变化并触发刷新。配置设置和特征标志的刷新相互独立。两者都由refresh方法触发，但特征标志的变化不会导致配置刷新，反之亦然。另外，如果未启用配置设置的刷新，特征标志仍然可以启用刷新。

config = load(endpoint=endpoint, credential=DefaultAzureCredential(), feature_flags_enabled=True, feature_flag_refresh_enabled=True)

...

config.refresh()

关键概念

故障排除

下一步

查看我们的Django和Flask示例，了解如何在Web应用程序中使用提供程序。

Django

Flask

贡献

本项目欢迎贡献和建议。大多数贡献都需要您同意贡献者许可协议（CLA），声明您有权并且实际上确实授予我们使用您的贡献的权利。有关详细信息，请访问https://cla.microsoft.com。

当您提交拉取请求时，CLA-bot会自动确定您是否需要提供CLA，并相应地装饰PR（例如，标签、注释）。只需遵循机器人提供的说明。您只需在整个使用我们的CLA的仓库中执行此操作一次。

本项目已采用Microsoft开源行为准则。有关更多信息，请参阅行为准则FAQ或通过opencode@microsoft.com联系，提出任何额外的问题或评论。

发布历史

1.3.0 (2024-09-09)

新增功能

• 添加了对副本之间的自动故障转移支持。
• 添加了对副本自动发现的支持。

1.2.0 (2024-05-24)

新增功能

• 启用加载具有feature_flag_enabled的特征标志
• 使用feature_flag_selectors选择要加载的特征标志
• 使用feature_flag_refresh_enabled启用/禁用特征标志刷新

已修复的错误

• 修复了由于返回配置副本而导致加载配置较慢的问题。

1.1.0 (2024-01-29)

新增功能

• 添加了用于从Azure App Configuration服务刷新配置的新API refresh。默认情况下，refresh可以每30秒检查一次指定的哨兵键的变化。如果检测到变化，则重新加载所有配置。可以通过传递SentinelKey的列表到refresh_on来设置哨兵键。
• 将新的选项on_refresh_success和on_refresh_failure回调添加到加载方法中。这些回调在刷新方法成功刷新配置或无法刷新配置时被调用。

已修复的错误

• 验证refresh_interval至少为1秒。

1.1.0b3 (2023-12-19)

新增功能

• 将on_refresh_success回调添加到加载方法中。此回调在刷新方法成功刷新配置时被调用。
• 添加了最小上线时间。这是提供程序在抛出错误之前尝试保持运行的最小时间。这是为了防止快速重启循环。

已修复的错误

• 修复了刷新计时器仅在找到变化后重置的问题。

其他更改

• 将类型SentinelKey重命名为WatchKey。

1.1.0b2 (2023-09-29)

新增功能

• 添加了对keyvault_credential、keyvault_client_configs和secret_resolver作为kwargs的支持，而不是使用AzureAppConfigurationKeyVaultOptions。

已修复的错误

• 修复了需要设置user_agent的问题。
• 修复了在刷新时相关性上下文信息错误的问题。

1.1.0b1 (2023-09-13)

新增功能

• 添加了用于从Azure App Configuration服务刷新配置的新API refresh。默认情况下，refresh可以每30秒检查一次指定的哨兵键的变化。如果检测到变化，则重新加载所有配置。可以通过传递SentinelKey的列表到refresh_on来设置哨兵键。
• 添加了对客户提供的用户代理前缀的支持。

其他更改

• 更新为使用 AZURE_APP_CONFIGURATION_TRACING_DISABLED 环境变量来禁用跟踪。
• 将重试的最大次数从默认的 3 次更改为 2 次。
• 将重试之间的最大退避时间从默认的 2 分钟更改为 1 分钟。
• 将 azure-core 的最低依赖性提升到 >=1.25.0

1.0.0 (2023-03-09)

重大变更

• 将 load_provider 重命名为 load
• 添加了 AzureAppConfigurationKeyVaultOptions 来接收一个端点到客户端 kwarg 的映射，而不是接收整个客户端。
• 移除了 AzureAppConfigurationKeyVaultOptions 的 secret_clients，应使用 client_configs。
• 为设置选择器添加了 key_filter 和 label_filter kwargs
• 将 trimmed_key_prefixes 重命名为 trim_prefixes

其他更改

• 将 EMPTY_LABEL 设为常量。即 "\0"

1.0.0b2 (2023-02-15)

新增功能

• 添加了异步支持
• 添加了 Mapping API 的缺失方法
• 将 load 方法属性设置为无序。

重大变更

• 更改了 load 的工作方式。将其从 AzureAppConfigurationProvider.load 移动到 load_provider。
• 移除了自定义 Key Vault 错误
• 移除了不需要的 repr 和 copy 方法。
• 所有功能标志都添加到它们自己的键中，并移除了前缀

已修复的错误

• 修复了在某些情况下无法设置 Key Vault 客户端的问题

其他更改

• 更新了方法文档
• 修复了使用 selector 而不是 selects 的 load 文档。
• 修复了 Readme 中的 CLI 链接。

1.0.0b1 (2022-10-13)

新的 Azure App Configuration Provider

提供对 Azure App Configuration SDK 之上的额外支持。启用

• 连接到 Azure App Configuration 存储
• 使用设置选择器选择多个密钥
• 当提供 AzureAppConfigurationKeyVaultOptions 时解析 Key Vault 引用

Azure App Configuration Provider 一次加载后返回一个键/值对字典，用于配置。

endpoint = "https://<your-store>.azconfig.io"
default_credential = DefaultAzureCredential()
config = AzureAppConfigurationProvider.load(
    endpoint=endpoint, credential=default_credential)
print(config["message"])