Snapshots python - Azure App Configuration Provider (#43690)

* Adding Snapshots

* fixing formatting

* Snapshot fixes + tests

* attempting to fix tests

* Update test_snapshots.py

* fixing recording and async tests

* Update assets.json

* format fixes

* fix live issue

* Apply suggestions from code review

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update snapshot_sample.py

* fix sample and test

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: mrm9084

sdk/appconfiguration/azure-appconfiguration-provider/assets.json

@@ -2,5 +2,5 @@
   "AssetsRepo": "Azure/azure-sdk-assets",
   "AssetsRepoPrefixPath": "python",
   "TagPrefix": "python/appconfiguration/azure-appconfiguration-provider",
-  "Tag": "python/appconfiguration/azure-appconfiguration-provider_ccc89e9eaa"
+  "Tag": "python/appconfiguration/azure-appconfiguration-provider_18bb5889fc"
 }

sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationproviderbase.py

@@ -127,6 +127,16 @@ def process_load_parameters(*args, **kwargs: Any) -> Dict[str, Any]:
     if kwargs.get("keyvault_credential") is not None and kwargs.get("secret_resolver") is not None:
         raise ValueError("A keyvault credential and secret resolver can't both be configured.")
 
+    # Validate feature flag selectors don't use snapshots
+    feature_flag_selectors = kwargs.get("feature_flag_selectors")
+    if feature_flag_selectors:
+        for selector in feature_flag_selectors:
+            if hasattr(selector, "snapshot_name") and selector.snapshot_name is not None:
+                raise ValueError(
+                    "snapshot_name cannot be used with feature_flag_selectors. "
+                    "Use snapshot_name with regular selects instead to load feature flags from snapshots."
+                )
+
     # Determine Key Vault usage
     uses_key_vault = (
         "keyvault_credential" in kwargs

sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_client_manager.py

@@ -17,6 +17,7 @@
     ConfigurationSetting,
     AzureAppConfigurationClient,
     FeatureFlagConfigurationSetting,
+    SnapshotComposition,
 )
 from ._client_manager_base import (
     _ConfigurationClientWrapperBase,
@@ -44,7 +45,7 @@ def from_credential(
         user_agent: str,
         retry_total: int,
         retry_backoff_max: int,
-        **kwargs
+        **kwargs,
     ) -> Self:
         """
         Creates a new instance of the _ConfigurationClientWrapper class, using the provided credential to authenticate
@@ -66,7 +67,7 @@ def from_credential(
                 user_agent=user_agent,
                 retry_total=retry_total,
                 retry_backoff_max=retry_backoff_max,
-                **kwargs
+                **kwargs,
             ),
         )
 
@@ -93,7 +94,7 @@ def from_connection_string(
                 user_agent=user_agent,
                 retry_total=retry_total,
                 retry_backoff_max=retry_backoff_max,
-                **kwargs
+                **kwargs,
             ),
         )
 
@@ -136,9 +137,20 @@ def _check_configuration_setting(
     def load_configuration_settings(self, selects: List[SettingSelector], **kwargs) -> List[ConfigurationSetting]:
         configuration_settings = []
         for select in selects:
-            configurations = self._client.list_configuration_settings(
-                key_filter=select.key_filter, label_filter=select.label_filter, tags_filter=select.tag_filters, **kwargs
-            )
+            if select.snapshot_name is not None:
+                # When loading from a snapshot, ignore key_filter, label_filter, and tag_filters
+                snapshot = self._client.get_snapshot(select.snapshot_name)
+                if snapshot.composition_type != SnapshotComposition.KEY:
+                    raise ValueError(f"Snapshot '{select.snapshot_name}' is not a key snapshot.")
+                configurations = self._client.list_configuration_settings(snapshot_name=select.snapshot_name, **kwargs)
+            else:
+                # Use traditional filtering when not loading from a snapshot
+                configurations = self._client.list_configuration_settings(
+                    key_filter=select.key_filter,
+                    label_filter=select.label_filter,
+                    tags_filter=select.tag_filters,
+                    **kwargs,
+                )
             for config in configurations:
                 if not isinstance(config, FeatureFlagConfigurationSetting):
                     # Feature flags are ignored when loaded by Selects, as they are selected from
@@ -154,11 +166,13 @@ def load_feature_flags(
         # Needs to be removed unknown keyword argument for list_configuration_settings
         kwargs.pop("sentinel_keys", None)
         for select in feature_flag_selectors:
+            # Handle None key_filter by converting to empty string
+            key_filter = select.key_filter if select.key_filter is not None else ""
             feature_flags = self._client.list_configuration_settings(
-                key_filter=FEATURE_FLAG_PREFIX + select.key_filter,
+                key_filter=FEATURE_FLAG_PREFIX + key_filter,
                 label_filter=select.label_filter,
                 tags_filter=select.tag_filters,
-                **kwargs
+                **kwargs,
             )
             for feature_flag in feature_flags:
                 if not isinstance(feature_flag, FeatureFlagConfigurationSetting):
@@ -261,7 +275,7 @@ def __init__(
         min_backoff_sec,
         max_backoff_sec,
         load_balancing_enabled,
-        **kwargs
+        **kwargs,
     ):
         super(ConfigurationClientManager, self).__init__(
             endpoint,
@@ -272,7 +286,7 @@ def __init__(
             min_backoff_sec,
             max_backoff_sec,
             load_balancing_enabled,
-            **kwargs
+            **kwargs,
         )
         self._original_connection_string = connection_string
         self._credential = credential
@@ -367,7 +381,7 @@ def refresh_clients(self):
                             self._user_agent,
                             self._retry_total,
                             self._retry_backoff_max,
-                            **self._args
+                            **self._args,
                         )
                     )
                 elif self._credential:
@@ -378,7 +392,7 @@ def refresh_clients(self):
                             self._user_agent,
                             self._retry_total,
                             self._retry_backoff_max,
-                            **self._args
+                            **self._args,
                         )
                     )
         self._next_update_time = time.time() + MINIMAL_CLIENT_REFRESH_INTERVAL

sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_models.py

@@ -6,6 +6,13 @@
 from typing import Optional, Callable, TYPE_CHECKING, Union, Awaitable, Mapping, Any, NamedTuple, List
 from ._constants import NULL_CHAR
 
+# Error message constants for SettingSelector validation
+_SNAPSHOT_PARAMETER_CONFLICT_ERROR = (
+    "Cannot specify both snapshot_name and {parameter}. "
+    "When using snapshots, all other filtering parameters are ignored."
+)
+_SELECTOR_REQUIRED_ERROR = "Either key_filter or snapshot_name must be specified."
+
 if TYPE_CHECKING:
     from azure.core.credentials import TokenCredential
     from azure.core.credentials_async import AsyncTokenCredential
@@ -43,21 +50,44 @@ class SettingSelector:
     """
     Selects a set of configuration settings from Azure App Configuration.
 
-    :keyword key_filter:A filter to select configuration settings and feature flags based on their keys.
+    :keyword key_filter: A filter to select configuration settings and feature flags based on their keys.
+     Cannot be used with snapshot_name.
     :type key_filter: str
     :keyword label_filter: A filter to select configuration settings and feature flags based on their labels. Default
-    is value is \0 i.e. (No Label) as seen in the portal.
+     value is \0 i.e. (No Label) as seen in the portal. Cannot be used with snapshot_name.
     :type label_filter: Optional[str]
     :keyword tag_filters: A filter to select configuration settings and feature flags based on their tags. This is a
-    list of strings that will be used to match tags on the configuration settings. Reserved characters (\\*, \\, ,)
-    must be escaped with backslash if they are part of the value. Tag filters must follow the format
-    "tagName=tagValue", for empty values use "tagName=" and for null values use "tagName=\\0".
+     list of strings that will be used to match tags on the configuration settings. Reserved characters (\\*, \\, ,)
+     must be escaped with backslash if they are part of the value. Tag filters must follow the format
+     "tagName=tagValue", for empty values use "tagName=" and for null values use "tagName=\\0".
+     Cannot be used with snapshot_name.
     :type tag_filters: Optional[List[str]]
+    :keyword snapshot_name: The name of the snapshot to load configuration settings from. When specified,
+     all configuration settings from the snapshot will be loaded. Cannot be used with key_filter, label_filter,
+     or tag_filters.
+    :type snapshot_name: Optional[str]
     """
 
     def __init__(
-        self, *, key_filter: str, label_filter: Optional[str] = NULL_CHAR, tag_filters: Optional[List[str]] = None
+        self,
+        *,
+        key_filter: Optional[str] = None,
+        label_filter: Optional[str] = NULL_CHAR,
+        tag_filters: Optional[List[str]] = None,
+        snapshot_name: Optional[str] = None,
     ):
+        if snapshot_name is not None:
+            # When using snapshots, no other filtering parameters should be specified
+            if key_filter is not None:
+                raise ValueError(_SNAPSHOT_PARAMETER_CONFLICT_ERROR.format(parameter="key_filter"))
+            if label_filter != NULL_CHAR:
+                raise ValueError(_SNAPSHOT_PARAMETER_CONFLICT_ERROR.format(parameter="label_filter"))
+            if tag_filters is not None:
+                raise ValueError(_SNAPSHOT_PARAMETER_CONFLICT_ERROR.format(parameter="tag_filters"))
+
+        if snapshot_name is None and key_filter is None:
+            raise ValueError(_SELECTOR_REQUIRED_ERROR)
+
         if tag_filters is not None:
             if not isinstance(tag_filters, list):
                 raise TypeError("tag_filters must be a list of strings.")
@@ -70,6 +100,7 @@ def __init__(
         self.key_filter = key_filter
         self.label_filter = label_filter
         self.tag_filters = tag_filters
+        self.snapshot_name = snapshot_name
 
 
 class WatchKey(NamedTuple):

sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_async_client_manager.py

@@ -15,6 +15,7 @@
 from azure.appconfiguration import (  # type:ignore # pylint:disable=no-name-in-module
     ConfigurationSetting,
     FeatureFlagConfigurationSetting,
+    SnapshotComposition,
 )
 from azure.appconfiguration.aio import AzureAppConfigurationClient
 from .._client_manager_base import (
@@ -46,7 +47,7 @@ def from_credential(
         user_agent: str,
         retry_total: int,
         retry_backoff_max: int,
-        **kwargs
+        **kwargs,
     ) -> Self:
         """
         Creates a new instance of the _AsyncConfigurationClientWrapper class, using the provided credential to
@@ -68,7 +69,7 @@ def from_credential(
                 user_agent=user_agent,
                 retry_total=retry_total,
                 retry_backoff_max=retry_backoff_max,
-                **kwargs
+                **kwargs,
             ),
         )
 
@@ -95,7 +96,7 @@ def from_connection_string(
                 user_agent=user_agent,
                 retry_total=retry_total,
                 retry_backoff_max=retry_backoff_max,
-                **kwargs
+                **kwargs,
             ),
         )
 
@@ -138,9 +139,20 @@ async def _check_configuration_setting(
     async def load_configuration_settings(self, selects: List[SettingSelector], **kwargs) -> List[ConfigurationSetting]:
         configuration_settings = []
         for select in selects:
-            configurations = self._client.list_configuration_settings(
-                key_filter=select.key_filter, label_filter=select.label_filter, tags_filter=select.tag_filters, **kwargs
-            )
+            if select.snapshot_name is not None:
+                # When loading from a snapshot, ignore key_filter, label_filter, and tag_filters
+                snapshot = await self._client.get_snapshot(select.snapshot_name)
+                if snapshot.composition_type != SnapshotComposition.KEY:
+                    raise ValueError(f"Snapshot '{select.snapshot_name}' is not a key snapshot.")
+                configurations = self._client.list_configuration_settings(snapshot_name=select.snapshot_name, **kwargs)
+            else:
+                # Use traditional filtering when not loading from a snapshot
+                configurations = self._client.list_configuration_settings(
+                    key_filter=select.key_filter,
+                    label_filter=select.label_filter,
+                    tags_filter=select.tag_filters,
+                    **kwargs,
+                )
             async for config in configurations:
                 if not isinstance(config, FeatureFlagConfigurationSetting):
                     # Feature flags are ignored when loaded by Selects, as they are selected from
@@ -156,11 +168,13 @@ async def load_feature_flags(
         # Needs to be removed unknown keyword argument for list_configuration_settings
         kwargs.pop("sentinel_keys", None)
         for select in feature_flag_selectors:
+            # Handle None key_filter by converting to empty string
+            key_filter = select.key_filter if select.key_filter is not None else ""
             feature_flags = self._client.list_configuration_settings(
-                key_filter=FEATURE_FLAG_PREFIX + select.key_filter,
+                key_filter=FEATURE_FLAG_PREFIX + key_filter,
                 label_filter=select.label_filter,
                 tags_filter=select.tag_filters,
-                **kwargs
+                **kwargs,
             )
             async for feature_flag in feature_flags:
                 if not isinstance(feature_flag, FeatureFlagConfigurationSetting):
@@ -265,7 +279,7 @@ def __init__(
         min_backoff_sec,
         max_backoff_sec,
         load_balancing_enabled,
-        **kwargs
+        **kwargs,
     ):
         super(AsyncConfigurationClientManager, self).__init__(
             endpoint,
@@ -276,7 +290,7 @@ def __init__(
             min_backoff_sec,
             max_backoff_sec,
             load_balancing_enabled,
-            **kwargs
+            **kwargs,
         )
         self._original_connection_string = connection_string
         self._credential = credential
@@ -373,7 +387,7 @@ async def refresh_clients(self):
                             self._user_agent,
                             self._retry_total,
                             self._retry_backoff_max,
-                            **self._args
+                            **self._args,
                         )
                     )
                 elif self._credential:
@@ -384,7 +398,7 @@ async def refresh_clients(self):
                             self._user_agent,
                             self._retry_total,
                             self._retry_backoff_max,
-                            **self._args
+                            **self._args,
                         )
                     )
         self._next_update_time = time.time() + MINIMAL_CLIENT_REFRESH_INTERVAL

sdk/appconfiguration/azure-appconfiguration-provider/samples/async_snapshot_sample.py

@@ -0,0 +1,55 @@
+# ------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for
+# license information.
+# -------------------------------------------------------------------------
+
+import asyncio
+from azure.appconfiguration.provider.aio import load
+from azure.appconfiguration.provider import SettingSelector
+from sample_utilities import get_client_modifications
+import os
+
+
+async def main():
+    kwargs = get_client_modifications()
+    connection_string = os.environ["APPCONFIGURATION_CONNECTION_STRING"]
+
+    # Loading configuration settings from a snapshot
+    # Note: The snapshot must already exist in your App Configuration store
+    snapshot_name = "my-snapshot-name"
+    snapshot_selects = [SettingSelector(snapshot_name=snapshot_name)]
+    config = await load(connection_string=connection_string, selects=snapshot_selects, **kwargs)
+
+    print("Configuration settings from snapshot:")
+    for key, value in config.items():
+        print(f"{key}: {value}")
+
+    # You can also combine snapshot-based selectors with regular selectors
+    # The snapshot settings and filtered settings will be merged, with later selectors taking precedence
+    mixed_selects = [
+        SettingSelector(snapshot_name=snapshot_name),  # Load all settings from snapshot
+        SettingSelector(key_filter="override.*", label_filter="prod"),  # Also load specific override settings
+    ]
+    config_mixed = await load(connection_string=connection_string, selects=mixed_selects, **kwargs)
+
+    print("\nMixed configuration (snapshot + filtered settings):")
+    for key, value in config_mixed.items():
+        print(f"{key}: {value}")
+
+    # Loading feature flags from a snapshot
+    # To load feature flags from a snapshot, include the snapshot selector in the 'selects' parameter and set feature_flag_enabled=True.
+    feature_flag_selects = [SettingSelector(snapshot_name=snapshot_name)]
+    config_with_flags = await load(
+        connection_string=connection_string,
+        selects=feature_flag_selects,
+        feature_flag_enabled=True,
+        **kwargs,
+    )
+    print(
+        f"\nConfiguration includes feature flags: {any(key.startswith('.appconfig.featureflag/') for key in config_with_flags.keys())}"
+    )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())