App Configuration - Snapshot references

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

@@ -27,6 +27,7 @@
 from ._constants import (
     FEATURE_MANAGEMENT_KEY,
     FEATURE_FLAG_KEY,
+    APP_CONFIG_SNAPSHOT_REF_MIME_PROFILE,
 )
 from ._azureappconfigurationproviderbase import (
     AzureAppConfigurationProviderBase,
@@ -299,7 +300,7 @@ def _attempt_refresh(self, client: ConfigurationClient, replica_count: int, is_f
 
             if settings_refreshed:
                 # Configuration Settings have been refreshed
-                processed_settings = self._process_configurations(configuration_settings)
+                processed_settings = self._process_configurations(configuration_settings, client)
 
             processed_settings = self._process_feature_flags(processed_settings, processed_feature_flags, feature_flags)
             self._dict = processed_settings
@@ -383,7 +384,7 @@ def _load_all(self, **kwargs: Any) -> None:
             try:
                 configuration_settings = client.load_configuration_settings(self._selects, headers=headers, **kwargs)
                 watched_settings = self._update_watched_settings(configuration_settings)
-                processed_settings = self._process_configurations(configuration_settings)
+                processed_settings = self._process_configurations(configuration_settings, client)
 
                 if self._feature_flag_enabled:
                     feature_flags: List[FeatureFlagConfigurationSetting] = client.load_feature_flags(
@@ -422,7 +423,7 @@ def _load_all(self, **kwargs: Any) -> None:
                 is_failover_request = True
         raise exception
 
-    def _process_configurations(self, configuration_settings: List[ConfigurationSetting]) -> Dict[str, Any]:
+    def _process_configurations(self, configuration_settings: List[ConfigurationSetting], client) -> Dict[str, Any]:
         # configuration_settings can contain duplicate keys, but they are in priority order, i.e. later settings take
         # precedence. Only process the settings with the highest priority (i.e. the last one in the list).
         unique_settings = self._deduplicate_settings(configuration_settings)
@@ -440,6 +441,31 @@ def _process_configurations(self, configuration_settings: List[ConfigurationSett
 
                 if self._feature_flag_refresh_enabled:
                     self._watched_feature_flags[(settings.key, settings.label)] = settings.etag
+            elif settings.content_type and APP_CONFIG_SNAPSHOT_REF_MIME_PROFILE in settings.content_type:
+                # Check if this is a snapshot reference
+
+                # Track snapshot reference usage for telemetry
+                self._tracing_context.uses_snapshot_reference = True
+                try:
+                    # Resolve the snapshot reference to actual settings
+                    snapshot_settings = client.resolve_snapshot_reference(settings)
+
+                    # Recursively process the resolved snapshot settings to handle feature flags,
+                    # configuration mapping
+                    snapshot_configuration_list = list(snapshot_settings.values())
+                    resolved_settings = self._process_configurations(snapshot_configuration_list, client)
+
+                    # Merge the resolved settings into our configuration
+                    configuration_settings_processed.update(resolved_settings)
+                except (ValueError, AzureError) as e:
+                    logger.warning(
+                        "Failed to resolve snapshot reference for key '%s' (label: '%s'): %s",
+                        settings.key,
+                        settings.label,
+                        str(e),
+                    )
+                    # Continue processing other settings even if snapshot resolution fails
+
             else:
                 key = self._process_key_name(settings)
                 value = self._process_key_value(settings)

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

@@ -39,6 +39,7 @@
     ALLOCATION_ID_KEY,
     APP_CONFIG_AI_MIME_PROFILE,
     APP_CONFIG_AICC_MIME_PROFILE,
+    APP_CONFIG_SNAPSHOT_REF_MIME_PROFILE,
     FEATURE_MANAGEMENT_KEY,
     FEATURE_FLAG_KEY,
 )
@@ -471,6 +472,8 @@ def _process_key_value_base(self, config: ConfigurationSetting) -> Union[str, Di
                     self._tracing_context.uses_ai_configuration = True
                 if APP_CONFIG_AICC_MIME_PROFILE in config.content_type:
                     self._tracing_context.uses_aicc_configuration = True
+                if APP_CONFIG_SNAPSHOT_REF_MIME_PROFILE in config.content_type:
+                    self._tracing_context.uses_snapshot_reference = True
                 return json.loads(config.value)
             except json.JSONDecodeError:
                 try:
@@ -481,6 +484,8 @@ def _process_key_value_base(self, config: ConfigurationSetting) -> Union[str, Di
                 except (json.JSONDecodeError, ValueError):
                     # If the value is not a valid JSON, treat it like regular string value
                     return config.value
+        elif config.content_type and APP_CONFIG_SNAPSHOT_REF_MIME_PROFILE in config.content_type:
+            self._tracing_context.uses_snapshot_reference = True
         return config.value
 
     def _process_feature_flags(

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

@@ -11,7 +11,7 @@
 from typing_extensions import Self
 from azure.core import MatchConditions
 from azure.core.tracing.decorator import distributed_trace
-from azure.core.exceptions import HttpResponseError
+from azure.core.exceptions import HttpResponseError, AzureError
 from azure.core.credentials import TokenCredential
 from azure.appconfiguration import (  # type:ignore # pylint:disable=no-name-in-module
     ConfigurationSetting,
@@ -28,6 +28,8 @@
 from ._models import SettingSelector
 from ._constants import FEATURE_FLAG_PREFIX
 from ._discovery import find_auto_failover_endpoints
+from ._snapshot_reference_parser import SnapshotReferenceParser
+from ._constants import APP_CONFIG_SNAPSHOT_REF_MIME_PROFILE
 
 
 @dataclass
@@ -261,6 +263,43 @@ def __enter__(self):
     def __exit__(self, *args):
         self._client.__exit__(*args)
 
+    def resolve_snapshot_reference(self, setting: ConfigurationSetting, **kwargs) -> Dict[str, ConfigurationSetting]:
+        """
+        Resolve a snapshot reference configuration setting to the actual snapshot data.
+
+        :param ConfigurationSetting setting: The snapshot reference configuration setting
+        :return: A dictionary of resolved configuration settings from the snapshot
+        :rtype: Dict[str, ConfigurationSetting]
+        :raises ValueError: When the setting is not a valid snapshot reference
+        """
+        if not setting.content_type or not APP_CONFIG_SNAPSHOT_REF_MIME_PROFILE in setting.content_type:
+            raise ValueError("Setting is not a snapshot reference")
+
+        try:
+            # Parse the snapshot reference
+            snapshot_name = SnapshotReferenceParser.parse(setting)
+
+            # Create a selector for the snapshot
+            snapshot_selector = SettingSelector(snapshot_name=snapshot_name)
+
+            # Use existing load_configuration_settings to load from snapshot
+            configurations = self.load_configuration_settings([snapshot_selector], **kwargs)
+
+            # Build a dictionary keyed by configuration key
+            snapshot_settings = {}
+            for config in configurations:
+                # Use lexicographic ordering - last wins for duplicate keys
+                snapshot_settings[config.key] = config
+
+            return snapshot_settings
+
+        except AzureError as e:
+            # Wrap Azure errors with more context
+            raise ValueError(
+                f"Failed to resolve snapshot reference for key '{setting.key}' "
+                f"(label: '{setting.label}'). Azure service error occurred."
+            ) from e
+
 
 class ConfigurationClientManager(ConfigurationClientManagerBase):  # pylint:disable=too-many-instance-attributes
     def __init__(

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

@@ -27,3 +27,11 @@
 # Mime profiles
 APP_CONFIG_AI_MIME_PROFILE = "https://azconfig.io/mime-profiles/ai/"
 APP_CONFIG_AICC_MIME_PROFILE = "https://azconfig.io/mime-profiles/ai/chat-completion"
+APP_CONFIG_SNAPSHOT_REF_MIME_PROFILE = "https://azconfig.io/mime-profiles/snapshot-ref"
+
+# Snapshot reference tracing key
+SNAPSHOT_REFERENCE_TAG = "UsesSnapshotReference=true"
+
+# Snapshot reference constants
+SNAPSHOT_REF_CONTENT_TYPE = 'application/json; profile="https://azconfig.io/mime-profiles/snapshot-ref"; charset=utf-8'
+SNAPSHOT_NAME_FIELD = "snapshot_name"

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

@@ -15,6 +15,7 @@
     KubernetesEnvironmentVariable,
     APP_CONFIG_AI_MIME_PROFILE,
     APP_CONFIG_AICC_MIME_PROFILE,
+    SNAPSHOT_REFERENCE_TAG,
 )
 
 # Feature flag filter names
@@ -86,6 +87,7 @@ def __init__(self, load_balancing_enabled: bool = False) -> None:
         self.uses_load_balancing = load_balancing_enabled
         self.uses_ai_configuration = False
         self.uses_aicc_configuration = False  # AI Chat Completion
+        self.uses_snapshot_reference = False
         self.uses_telemetry = False
         self.uses_seed = False
         self.max_variants: Optional[int] = None
@@ -218,6 +220,9 @@ def update_correlation_context_header(
         if self.is_failover_request:
             tags.append(FAILOVER_TAG)
 
+        if self.uses_snapshot_reference:
+            tags.append(SNAPSHOT_REFERENCE_TAG)
+
         # Build the correlation context string
         context_parts: List[str] = []
 

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

@@ -0,0 +1,77 @@
+# ------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for
+# license information.
+# -------------------------------------------------------------------------
+
+import json
+from azure.appconfiguration import ConfigurationSetting  # type: ignore
+from ._constants import SNAPSHOT_NAME_FIELD
+
+
+class SnapshotReferenceParser:
+    """
+    Parser for snapshot reference configuration settings.
+    """
+
+    @staticmethod
+    def parse(setting: ConfigurationSetting) -> str:
+        """
+        Parse a snapshot reference from a configuration setting containing snapshot reference JSON.
+
+        :param ConfigurationSetting setting: The configuration setting containing the snapshot reference JSON
+        :return: The snapshot name extracted from the reference
+        :rtype: str
+        :raises ValueError: When the setting is None
+        :raises ValueError: When the setting contains invalid JSON, invalid snapshot reference format,
+                           or empty/whitespace snapshot name
+        """
+        if setting is None:
+            raise ValueError("Setting cannot be None")
+
+        if not setting.value or setting.value.strip() == "":
+            raise ValueError(
+                f"Invalid snapshot reference format for key '{setting.key}' "
+                f"(label: '{setting.label}'). Value cannot be empty."
+            )
+
+        try:
+            # Parse the JSON content
+            json_content = json.loads(setting.value)
+
+            if not isinstance(json_content, dict):
+                raise ValueError(
+                    f"Invalid snapshot reference format for key '{setting.key}' "
+                    f"(label: '{setting.label}'). Expected JSON object."
+                )
+
+            # Extract the snapshot name
+            snapshot_name = json_content.get(SNAPSHOT_NAME_FIELD)
+
+            if snapshot_name is None:
+                raise ValueError(
+                    f"Invalid snapshot reference format for key '{setting.key}' "
+                    f"(label: '{setting.label}'). The '{SNAPSHOT_NAME_FIELD}' "
+                    f"property is required."
+                )
+
+            if not isinstance(snapshot_name, str):
+                raise ValueError(
+                    f"Invalid snapshot reference format for key '{setting.key}' "
+                    f"(label: '{setting.label}'). The '{SNAPSHOT_NAME_FIELD}' "
+                    f"property must be a string value, but found {type(snapshot_name).__name__}."
+                )
+
+            if not snapshot_name.strip():
+                raise ValueError(
+                    f"Invalid snapshot reference format for key '{setting.key}' "
+                    f"(label: '{setting.label}'). Snapshot name cannot be empty or whitespace."
+                )
+
+            return snapshot_name.strip()
+
+        except json.JSONDecodeError as json_ex:
+            raise ValueError(
+                f"Invalid snapshot reference format for key '{setting.key}' "
+                f"(label: '{setting.label}'). Invalid JSON format."
+            ) from json_ex

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

@@ -11,7 +11,7 @@
 from typing_extensions import Self
 from azure.core import MatchConditions
 from azure.core.tracing.decorator import distributed_trace
-from azure.core.exceptions import HttpResponseError
+from azure.core.exceptions import HttpResponseError, AzureError
 from azure.appconfiguration import (  # type:ignore # pylint:disable=no-name-in-module
     ConfigurationSetting,
     FeatureFlagConfigurationSetting,
@@ -26,6 +26,8 @@
 )
 from .._models import SettingSelector
 from .._constants import FEATURE_FLAG_PREFIX
+from .._snapshot_reference_parser import SnapshotReferenceParser
+from .._constants import APP_CONFIG_SNAPSHOT_REF_MIME_PROFILE
 from ._async_discovery import find_auto_failover_endpoints
 
 if TYPE_CHECKING:
@@ -265,6 +267,45 @@ async def __aenter__(self):
     async def __aexit__(self, *args):
         await self._client.__aexit__(*args)
 
+    async def resolve_snapshot_reference(
+        self, setting: ConfigurationSetting, **kwargs
+    ) -> Dict[str, ConfigurationSetting]:
+        """
+        Resolve a snapshot reference configuration setting to the actual snapshot data.
+
+        :param ConfigurationSetting setting: The snapshot reference configuration setting
+        :return: A dictionary of resolved configuration settings from the snapshot
+        :rtype: Dict[str, ConfigurationSetting]
+        :raises ValueError: When the setting is not a valid snapshot reference
+        """
+        if not setting.content_type or not APP_CONFIG_SNAPSHOT_REF_MIME_PROFILE in setting.content_type:
+            raise ValueError("Setting is not a snapshot reference")
+
+        try:
+            # Parse the snapshot reference
+            snapshot_name = SnapshotReferenceParser.parse(setting)
+
+            # Create a selector for the snapshot
+            snapshot_selector = SettingSelector(snapshot_name=snapshot_name)
+
+            # Use existing load_configuration_settings to load from snapshot
+            configurations = await self.load_configuration_settings([snapshot_selector], **kwargs)
+
+            # Build a dictionary keyed by configuration key
+            snapshot_settings = {}
+            for config in configurations:
+                # Use lexicographic ordering - last wins for duplicate keys
+                snapshot_settings[config.key] = config
+
+            return snapshot_settings
+
+        except AzureError as e:
+            # Wrap Azure errors with more context
+            raise ValueError(
+                f"Failed to resolve snapshot reference for key '{setting.key}' "
+                f"(label: '{setting.label}'). Azure service error occurred."
+            ) from e
+
 
 class AsyncConfigurationClientManager(ConfigurationClientManagerBase):  # pylint:disable=too-many-instance-attributes
     def __init__(