App Config Provider - Update Locking Logic (#33433)

mrm9084 · web-flow · commit bcd3840fbe98 · 2023-12-15T17:21:34.000-08:00
* Update to locking logic

* Updated from comments

* Formatting

* Fixing Line Length

* Update _azureappconfigurationproviderasync.py

* Update _azureappconfigurationproviderasync.py

* Updating locking

* fix await

* Fixing lock check

* Reverting

* Update _azureappconfigurationprovider.py

* Update _azureappconfigurationproviderasync.py

* pylint

* Update _azureappconfigurationproviderasync.py

* ): # pylint: disable= consider-using-with

* Update _azureappconfigurationprovider.py

* formatting

* Removed redundant return

* Fixed needs refresh, removed old prop

* missing word

* also missing
diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationprovider.py b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationprovider.py
@@ -3,6 +3,7 @@
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
 # -------------------------------------------------------------------------
+import copy
 import os
 import json
 import random
@@ -438,63 +439,65 @@ def __init__(self, **kwargs) -> None:
             or self._secret_resolver is not None
         )
         self._update_lock = Lock()
+        self._refresh_lock = Lock()
 
     def refresh(self, **kwargs) -> None:
         if not self._refresh_on:
             logging.debug("Refresh called but no refresh options set.")
             return
-
         if not self._refresh_timer.needs_refresh():
             logging.debug("Refresh called but refresh interval not elapsed.")
             return
+        if not self._refresh_lock.acquire(blocking=False):  # pylint: disable= consider-using-with
+            logging.debug("Refresh called but refresh already in progress.")
+            return
         success = False
         need_refresh = False
         try:
-            with self._update_lock:
-                updated_sentinel_keys = dict(self._refresh_on)
-                headers = _get_headers("Watch", uses_key_vault=self._uses_key_vault, **kwargs)
-                for (key, label), etag in updated_sentinel_keys.items():
-                    try:
-                        updated_sentinel = self._client.get_configuration_setting(
-                            key=key,
-                            label=label,
-                            etag=etag,
-                            match_condition=MatchConditions.IfModified,
-                            headers=headers,
-                            **kwargs
+            updated_sentinel_keys = dict(self._refresh_on)
+            headers = _get_headers("Watch", uses_key_vault=self._uses_key_vault, **kwargs)
+            for (key, label), etag in updated_sentinel_keys.items():
+                try:
+                    updated_sentinel = self._client.get_configuration_setting(
+                        key=key,
+                        label=label,
+                        etag=etag,
+                        match_condition=MatchConditions.IfModified,
+                        headers=headers,
+                        **kwargs
+                    )
+                    if updated_sentinel is not None:
+                        logging.debug(
+                            "Refresh all triggered by key: %s label %s.",
+                            key,
+                            label,
                         )
-                        if updated_sentinel is not None:
-                            logging.debug(
-                                "Refresh all triggered by key: %s label %s.",
-                                key,
-                                label,
-                            )
+                        need_refresh = True
+
+                        updated_sentinel_keys[(key, label)] = updated_sentinel.etag
+                except HttpResponseError as e:
+                    if e.status_code == 404:
+                        if etag is not None:
+                            # If the sentinel is not found, it means the key/label was deleted, so we should refresh
+                            logging.debug("Refresh all triggered by key: %s label %s.", key, label)
                             need_refresh = True
-
-                            updated_sentinel_keys[(key, label)] = updated_sentinel.etag
-                    except HttpResponseError as e:
-                        if e.status_code == 404:
-                            if etag is not None:
-                                # If the sentinel is not found, it means the key/label was deleted, so we should refresh
-                                logging.debug("Refresh all triggered by key: %s label %s.", key, label)
-                                need_refresh = True
-                                updated_sentinel_keys[(key, label)] = None
-                        else:
-                            raise e
-                # Need to only update once, no matter how many sentinels are updated
-                if need_refresh:
-                    self._load_all(headers=headers, sentinel_keys=updated_sentinel_keys, **kwargs)
-                # Even if we don't need to refresh, we should reset the timer
-                self._refresh_timer.reset()
-                success = True
-                return
+                            updated_sentinel_keys[(key, label)] = None
+                    else:
+                        raise e
+            # Need to only update once, no matter how many sentinels are updated
+            if need_refresh:
+                self._load_all(headers=headers, sentinel_keys=updated_sentinel_keys, **kwargs)
+            # Even if we don't need to refresh, we should reset the timer
+            self._refresh_timer.reset()
+            success = True
         except (ServiceRequestError, ServiceResponseError, HttpResponseError) as e:
             # If we get an error we should retry sooner than the next refresh interval
             if self._on_refresh_error:
                 self._on_refresh_error(e)
                 return
             raise
         finally:
+            self._refresh_lock.release()
             if not success:
                 self._refresh_timer.backoff()
             elif need_refresh and self._on_refresh_success:
@@ -523,7 +526,8 @@ def _load_all(self, **kwargs):
                 if (config.key, config.label) in self._refresh_on:
                     sentinel_keys[(config.key, config.label)] = config.etag
         self._refresh_on = sentinel_keys
-        self._dict = configuration_settings
+        with self._update_lock:
+            self._dict = configuration_settings
 
     def _process_key_name(self, config):
         trimmed_key = config.key
@@ -576,42 +580,43 @@ def keys(self) -> Iterable[str]:
         :rtype: Iterable[str]
         """
         with self._update_lock:
-            return self._dict.keys()
+            return list(self._dict.keys())
 
-    def items(self) -> Iterable[Tuple[str, str]]:
+    def items(self) -> Iterable[Tuple[str, Union[str, JSON]]]:
         """
-        Returns a list of key-value pairs loaded from Azure App Configuration. Any values that are Key Vault references
-        will be resolved.
+        Returns a set-like object of key-value pairs loaded from Azure App Configuration. Any values that are Key Vault
+         references will be resolved.
 
-        :return: A list of key-value pairs loaded from Azure App Configuration.
-        :rtype: Iterable[Tuple[str, str]]
+        :return: A set-like object of key-value pairs loaded from Azure App Configuration.
+        :rtype: Iterable[Tuple[str, Union[str, JSON]]]
         """
         with self._update_lock:
-            return self._dict.items()
+            return copy.deepcopy(self._dict.items())
 
-    def values(self) -> Iterable[str]:
+    def values(self) -> Iterable[Union[str, JSON]]:
         """
         Returns a list of values loaded from Azure App Configuration. Any values that are Key Vault references will be
         resolved.
 
-        :return: A list of values loaded from Azure App Configuration.
-        :rtype: Iterable[str]
+        :return: A list of values loaded from Azure App Configuration. The values are either Strings or JSON objects,
+        based on there content type.
+        :rtype: Iterable[[str], [JSON]]
         """
         with self._update_lock:
-            return self._dict.values()
+            return copy.deepcopy(list((self._dict.values())))
 
-    def get(self, key: str, default: Optional[str] = None) -> str:
+    def get(self, key: str, default: Optional[str] = None) -> Union[str, JSON]:
         """
         Returns the value of the specified key. If the key does not exist, returns the default value.
 
         :param str key: The key of the value to get.
         :param default: The default value to return.
         :type: str or None
         :return: The value of the specified key.
-        :rtype: str
+        :rtype: Union[str, JSON]
         """
         with self._update_lock:
-            return self._dict.get(key, default)
+            return copy.deepcopy(self._dict.get(key, default))
 
     def __eq__(self, other: Any) -> bool:
         if not isinstance(other, AzureAppConfigurationProvider):
diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_azureappconfigurationproviderasync.py b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_azureappconfigurationproviderasync.py
@@ -4,8 +4,9 @@
 # license information.
 # -------------------------------------------------------------------------
 import json
+import copy
+from threading import Lock
 import datetime
-from asyncio.locks import Lock
 import logging
 from typing import (
     Any,
@@ -330,58 +331,61 @@ def __init__(self, **kwargs) -> None:
             or self._secret_resolver is not None
         )
         self._update_lock = Lock()
+        self._refresh_lock = Lock()
 
     async def refresh(self, **kwargs) -> None:
         if not self._refresh_on:
             logging.debug("Refresh called but no refresh options set.")
             return
+        if not self._refresh_timer.needs_refresh():
+            logging.debug("Refresh called but refresh interval not elapsed.")
+            return
+        if not self._refresh_lock.acquire(blocking=False):  # pylint: disable= consider-using-with
+            logging.debug("Refresh called but refresh already in progress.")
+            return
         success = False
         need_refresh = False
         try:
-            async with self._update_lock:
-                if not self._refresh_timer.needs_refresh():
-                    logging.debug("Refresh called but refresh interval not elapsed.")
-                    return
-                updated_sentinel_keys = dict(self._refresh_on)
-                headers = _get_headers("Watch", uses_key_vault=self._uses_key_vault, **kwargs)
-                for (key, label), etag in updated_sentinel_keys.items():
-                    try:
-                        updated_sentinel = await self._client.get_configuration_setting(
-                            key=key,
-                            label=label,
-                            etag=etag,
-                            match_condition=MatchConditions.IfModified,
-                            headers=headers,
-                            **kwargs
-                        )
-                        if updated_sentinel is not None:
+            updated_sentinel_keys = dict(self._refresh_on)
+            headers = _get_headers("Watch", uses_key_vault=self._uses_key_vault, **kwargs)
+            for (key, label), etag in updated_sentinel_keys.items():
+                try:
+                    updated_sentinel = await self._client.get_configuration_setting(
+                        key=key,
+                        label=label,
+                        etag=etag,
+                        match_condition=MatchConditions.IfModified,
+                        headers=headers,
+                        **kwargs
+                    )
+                    if updated_sentinel is not None:
+                        logging.debug("Refresh all triggered by key: %s label %s.", key, label)
+                        need_refresh = True
+
+                        updated_sentinel_keys[(key, label)] = updated_sentinel.etag
+                except HttpResponseError as e:
+                    if e.status_code == 404:
+                        if etag is not None:
+                            # If the sentinel is not found, it means the key/label was deleted, so we should refresh
                             logging.debug("Refresh all triggered by key: %s label %s.", key, label)
                             need_refresh = True
-
-                            updated_sentinel_keys[(key, label)] = updated_sentinel.etag
-                    except HttpResponseError as e:
-                        if e.status_code == 404:
-                            if etag is not None:
-                                # If the sentinel is not found, it means the key/label was deleted, so we should refresh
-                                logging.debug("Refresh all triggered by key: %s label %s.", key, label)
-                                need_refresh = True
-                                updated_sentinel_keys[(key, label)] = None
-                        else:
-                            raise e
-                # Need to only update once, no matter how many sentinels are updated
-                if need_refresh:
-                    await self._load_all(headers=headers, sentinel_keys=updated_sentinel_keys, **kwargs)
-                # Even if we don't need to refresh, we should reset the timer
-                self._refresh_timer.reset()
-                success = True
-                return
+                            updated_sentinel_keys[(key, label)] = None
+                    else:
+                        raise e
+            # Need to only update once, no matter how many sentinels are updated
+            if need_refresh:
+                await self._load_all(headers=headers, sentinel_keys=updated_sentinel_keys, **kwargs)
+            # Even if we don't need to refresh, we should reset the timer
+            self._refresh_timer.reset()
+            success = True
         except (ServiceRequestError, ServiceResponseError, HttpResponseError) as e:
             # If we get an error we should retry sooner than the next refresh interval
             if self._on_refresh_error:
                 await self._on_refresh_error(e)
                 return
             raise
         finally:
+            self._refresh_lock.release()
             if not success:
                 self._refresh_timer.backoff()
             elif need_refresh and self._on_refresh_success:
@@ -463,39 +467,40 @@ def keys(self) -> Iterable[str]:
         :return: A list of keys loaded from Azure App Configuration.
         :rtype: Iterable[str]
         """
-        return self._dict.keys()
+        return list(self._dict.keys())
 
-    def items(self) -> Iterable[Tuple[str, str]]:
+    def items(self) -> Iterable[Tuple[str, Union[str, JSON]]]:
         """
-        Returns a list of key-value pairs loaded from Azure App Configuration. Any values that are Key Vault references
-        will be resolved.
+        Returns a set-like object of key-value pairs loaded from Azure App Configuration. Any values that are Key Vault
+         references will be resolved.
 
-        :return: A list of key-value pairs loaded from Azure App Configuration.
-        :rtype: Iterable[Tuple[str, str]]
+        :return: A set-like object of key-value pairs loaded from Azure App Configuration.
+        :rtype: Iterable[Tuple[str, Union[str, JSON]]]
         """
-        return self._dict.items()
+        return copy.deepcopy(self._dict.items())
 
-    def values(self) -> Iterable[str]:
+    def values(self) -> Iterable[Union[str, JSON]]:
         """
         Returns a list of values loaded from Azure App Configuration. Any values that are Key Vault references will be
         resolved.
 
-        :return: A list of values loaded from Azure App Configuration.
-        :rtype: Iterable[str]
+        :return: A list of values loaded from Azure App Configuration. The values are either Strings or JSON objects,
+        based on there content type.
+        :rtype: Iterable[[str], [JSON]]
         """
-        return self._dict.values()
+        return copy.deepcopy(list((self._dict.values())))
 
-    def get(self, key: str, default: Optional[str] = None) -> str:
+    def get(self, key: str, default: Optional[str] = None) -> Union[str, JSON]:
         """
         Returns the value of the specified key. If the key does not exist, returns the default value.
 
         :param str key: The key of the value to get.
         :param default: The default value to return.
         :type: str or None
         :return: The value of the specified key.
-        :rtype: str
+        :rtype: Union[str, JSON]
         """
-        return self._dict.get(key, default)
+        return copy.deepcopy(self._dict.get(key, default))
 
     def __eq__(self, other: Any) -> bool:
         if not isinstance(other, AzureAppConfigurationProvider):
