App config provider fix my py next (#33905)

* Adding mypy ignores

* MyPy fixes

* format fix

* Fixed PyLint

* fixing refresh on

* fixing mypy errors

* updated for mypy changes

* Fixing MyPy issue

* Fixes pylint issue

* Fixed check

---------

Co-authored-by: Krista Pratico <[email protected]>

Author: mrm9084

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

@@ -15,16 +15,20 @@
     Any,
     Callable,
     Dict,
-    Iterable,
     Mapping,
     Optional,
     overload,
     List,
     Tuple,
     TYPE_CHECKING,
     Union,
+    Iterator,
+    KeysView,
+    ItemsView,
+    ValuesView,
+    TypeVar,
 )
-from azure.appconfiguration import (  # pylint:disable=no-name-in-module
+from azure.appconfiguration import (  # type:ignore # pylint:disable=no-name-in-module
     AzureAppConfigurationClient,
     FeatureFlagConfigurationSetting,
     SecretReferenceConfigurationSetting,
@@ -49,8 +53,8 @@
 if TYPE_CHECKING:
     from azure.core.credentials import TokenCredential
 
-JSON = Union[str, Mapping[str, Any]]  # pylint: disable=unsubscriptable-object
-
+JSON = Mapping[str, Any]  # pylint: disable=unsubscriptable-object
+_T = TypeVar("_T")
 logger = logging.getLogger(__name__)
 
 min_uptime = 5
@@ -63,6 +67,9 @@ def load(
     *,
     selects: Optional[List[SettingSelector]] = None,
     trim_prefixes: Optional[List[str]] = None,
+    keyvault_credential: Optional["TokenCredential"] = None,
+    keyvault_client_configs: Optional[Mapping[str, JSON]] = None,
+    secret_resolver: Optional[Callable[[str], str]] = None,
     key_vault_options: Optional[AzureAppConfigurationKeyVaultOptions] = None,
     refresh_on: Optional[List[Tuple[str, str]]] = None,
     refresh_interval: int = 30,
@@ -109,6 +116,9 @@ def load(
     connection_string: str,
     selects: Optional[List[SettingSelector]] = None,
     trim_prefixes: Optional[List[str]] = None,
+    keyvault_credential: Optional["TokenCredential"] = None,
+    keyvault_client_configs: Optional[Mapping[str, JSON]] = None,
+    secret_resolver: Optional[Callable[[str], str]] = None,
     key_vault_options: Optional[AzureAppConfigurationKeyVaultOptions] = None,
     refresh_on: Optional[List[Tuple[str, str]]] = None,
     refresh_interval: int = 30,
@@ -201,8 +211,8 @@ def load(*args, **kwargs) -> "AzureAppConfigurationProvider":
     for (key, label), etag in provider._refresh_on.items():
         if not etag:
             try:
-                sentinel = provider._client.get_configuration_setting(key, label, headers=headers)
-                provider._refresh_on[(key, label)] = sentinel.etag
+                sentinel = provider._client.get_configuration_setting(key, label, headers=headers)  # type:ignore
+                provider._refresh_on[(key, label)] = sentinel.etag  # type:ignore
             except HttpResponseError as e:
                 if e.status_code == 404:
                     # If the sentinel is not found a refresh should be triggered when it is created.
@@ -279,15 +289,17 @@ def _buildprovider(
             **kwargs
         )
         return provider
-    provider._client = AzureAppConfigurationClient(
-        endpoint,
-        credential,
-        user_agent=user_agent,
-        retry_total=retry_total,
-        retry_backoff_max=retry_backoff_max,
-        **kwargs
-    )
-    return provider
+    if endpoint is not None and credential is not None:
+        provider._client = AzureAppConfigurationClient(
+            endpoint,
+            credential,
+            user_agent=user_agent,
+            retry_total=retry_total,
+            retry_backoff_max=retry_backoff_max,
+            **kwargs
+        )
+        return provider
+    raise ValueError("Please pass either endpoint and credential, or a connection string.")
 
 
 def _resolve_keyvault_reference(
@@ -320,7 +332,9 @@ def _resolve_keyvault_reference(
         provider._secret_clients[vault_url] = referenced_client
 
     if referenced_client:
-        return referenced_client.get_secret(keyvault_identifier.name, version=keyvault_identifier.version).value
+        secret_value = referenced_client.get_secret(keyvault_identifier.name, version=keyvault_identifier.version).value
+        if secret_value is not None:
+            return secret_value
 
     if provider._secret_resolver:
         return provider._secret_resolver(config.secret_id)
@@ -352,7 +366,7 @@ def _is_json_content_type(content_type: str) -> bool:
 
 def _build_sentinel(setting: Union[str, Tuple[str, str]]) -> Tuple[str, str]:
     try:
-        key, label = setting
+        key, label = setting  # type:ignore
     except IndexError:
         key = setting
         label = EMPTY_LABEL
@@ -417,18 +431,17 @@ class AzureAppConfigurationProvider(Mapping[str, Union[str, JSON]]):  # pylint:
 
     def __init__(self, **kwargs) -> None:
         self._dict: Dict[str, str] = {}
-        self._trim_prefixes: List[str] = []
         self._client: Optional[AzureAppConfigurationClient] = None
         self._secret_clients: Dict[str, SecretClient] = {}
         self._selects: List[SettingSelector] = kwargs.pop(
             "selects", [SettingSelector(key_filter="*", label_filter=EMPTY_LABEL)]
         )
 
         trim_prefixes: List[str] = kwargs.pop("trim_prefixes", [])
-        self._trim_prefixes = sorted(trim_prefixes, key=len, reverse=True)
+        self._trim_prefixes: List[str] = sorted(trim_prefixes, key=len, reverse=True)
 
         refresh_on: List[Tuple[str, str]] = kwargs.pop("refresh_on", None) or []
-        self._refresh_on: Mapping[Tuple[str, str] : Optional[str]] = {_build_sentinel(s): None for s in refresh_on}
+        self._refresh_on: Mapping[Tuple[str, str], Optional[str]] = {_build_sentinel(s): None for s in refresh_on}
         self._refresh_timer: _RefreshTimer = _RefreshTimer(**kwargs)
         self._on_refresh_success: Optional[Callable] = kwargs.pop("on_refresh_success", None)
         self._on_refresh_error: Optional[Callable[[Exception], None]] = kwargs.pop("on_refresh_error", None)
@@ -460,7 +473,7 @@ def refresh(self, **kwargs) -> None:
             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(
+                    updated_sentinel = self._client.get_configuration_setting(  # type:ignore
                         key=key,
                         label=label,
                         etag=etag,
@@ -554,14 +567,14 @@ def _process_key_value(self, config):
                 return config.value
         return config.value
 
-    def __getitem__(self, key: str) -> str:
+    def __getitem__(self, key: str) -> Any:
         # pylint:disable=docstring-missing-param,docstring-missing-return,docstring-missing-rtype
         """
         Returns the value of the specified key.
         """
         return self._dict[key]
 
-    def __iter__(self) -> Iterable[str]:
+    def __iter__(self) -> Iterator[str]:
         return self._dict.__iter__()
 
     def __len__(self) -> int:
@@ -574,40 +587,48 @@ def __contains__(self, __x: object) -> bool:
         """
         return self._dict.__contains__(__x)
 
-    def keys(self) -> Iterable[str]:
+    def keys(self) -> KeysView[str]:
         """
         Returns a list of keys loaded from Azure App Configuration.
 
         :return: A list of keys loaded from Azure App Configuration.
-        :rtype: Iterable[str]
+        :rtype: KeysView[str]
         """
         with self._update_lock:
-            return list(self._dict.keys())
+            return copy.deepcopy(self._dict).keys()
 
-    def items(self) -> Iterable[Tuple[str, Union[str, JSON]]]:
+    def items(self) -> ItemsView[str, Union[str, Mapping[str, Any]]]:
         """
         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 set-like object of key-value pairs loaded from Azure App Configuration.
-        :rtype: Iterable[Tuple[str, Union[str, JSON]]]
+        :rtype: ItemsView[str, Union[str, Mapping[str, Any]]]
         """
         with self._update_lock:
-            return copy.deepcopy(self._dict.items())
+            return copy.deepcopy(self._dict).items()
 
-    def values(self) -> Iterable[Union[str, JSON]]:
+    def values(self) -> ValuesView[Union[str, Mapping[str, Any]]]:
         """
         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. The values are either Strings or JSON objects,
-        based on there content type.
-        :rtype: Iterable[[str], [JSON]]
+         based on there content type.
+        :rtype: ValuesView[Union[str, Mapping[str, Any]]]
         """
         with self._update_lock:
-            return copy.deepcopy(list((self._dict.values())))
+            return copy.deepcopy((self._dict)).values()
+
+    @overload
+    def get(self, key: str, default: None = None) -> Union[str, JSON, None]:
+        ...
+
+    @overload
+    def get(self, key: str, default: Union[str, JSON, _T]) -> Union[str, JSON, _T]:  # pylint: disable=signature-differs
+        ...
 
-    def get(self, key: str, default: Optional[str] = None) -> Union[str, JSON]:
+    def get(self, key: str, default: Optional[Union[str, JSON, _T]] = None) -> Union[str, JSON, _T, None]:
         """
         Returns the value of the specified key. If the key does not exist, returns the default value.
 
@@ -618,7 +639,7 @@ def get(self, key: str, default: Optional[str] = None) -> Union[str, JSON]:
         :rtype: Union[str, JSON]
         """
         with self._update_lock:
-            return copy.deepcopy(self._dict.get(key, default))
+            return copy.deepcopy(self._dict).get(key, default)
 
     def __eq__(self, other: Any) -> bool:
         if not isinstance(other, AzureAppConfigurationProvider):
@@ -640,15 +661,15 @@ def close(self) -> None:
         """
         for client in self._secret_clients.values():
             client.close()
-        self._client.close()
+        self._client.close()  # type: ignore
 
     def __enter__(self) -> "AzureAppConfigurationProvider":
-        self._client.__enter__()
+        self._client.__enter__()  # type: ignore
         for client in self._secret_clients.values():
             client.__enter__()
         return self
 
     def __exit__(self, *args) -> None:
-        self._client.__exit__(*args)
+        self._client.__exit__(*args)  # type: ignore
         for client in self._secret_clients.values():
             client.__exit__()