[AppConfig] mypy fixes (Azure#18858)

closes Azure#18540

Author: seankane-msft

eng/tox/mypy_hard_failure_packages.py

@@ -15,5 +15,6 @@
   "azure-ai-formrecognizer",
   "azure-ai-metricsadvisor",
   "azure-eventgrid",
+  "azure-appconfiguration",
   "azure-data-tables",
 ]

sdk/appconfiguration/azure-appconfiguration/azure/__init__.py

@@ -1 +1 @@
-__path__ = __import__("pkgutil").extend_path(__path__, __name__)
+__path__ = __import__("pkgutil").extend_path(__path__, __name__)  # type: ignore

sdk/appconfiguration/azure-appconfiguration/azure/appconfiguration/_azure_appconfiguration_client.py

@@ -4,7 +4,7 @@
 # license information.
 # -------------------------------------------------------------------------
 import binascii
-from typing import Optional, Any
+from typing import Optional, Any, Mapping, Union
 from requests.structures import CaseInsensitiveDict
 from azure.core import MatchConditions
 from azure.core.pipeline import Pipeline
@@ -83,8 +83,6 @@ def __init__(self, base_url, credential, **kwargs):
         )
         self._sync_token_policy = SyncTokenPolicy()
 
-        self._sync_token_policy = None
-
         pipeline = kwargs.get("pipeline")
 
         if pipeline is None:
@@ -204,7 +202,7 @@ def list_configuration_settings(
         error_map = {401: ClientAuthenticationError}
 
         try:
-            return self._impl.get_key_values(
+            return self._impl.get_key_values(  # type: ignore
                 label=label_filter,
                 key=key_filter,
                 select=select,
@@ -223,12 +221,12 @@ def list_configuration_settings(
     @distributed_trace
     def get_configuration_setting(
         self,
-        key,
-        label=None,
-        etag="*",
-        match_condition=MatchConditions.Unconditionally,
-        **kwargs
-    ):  # type: (str, Optional[str], Optional[str], Optional[MatchConditions], **Any) -> ConfigurationSetting
+        key,  # type: str
+        label=None,  # type: Optional[str]
+        etag="*",  # type: Optional[str]
+        match_condition=MatchConditions.Unconditionally,  # type: Optional[MatchConditions]
+        **kwargs  # type: Any
+    ):  # type: (...) -> Union[None, ConfigurationSetting]
 
         """Get the matched ConfigurationSetting from Azure App Configuration service
 
@@ -310,12 +308,12 @@ def add_configuration_setting(self, configuration_setting, **kwargs):
             added_config_setting = client.add_configuration_setting(config_setting)
         """
         key_value = configuration_setting._to_generated()
-        custom_headers = CaseInsensitiveDict(kwargs.get("headers"))
+        custom_headers = CaseInsensitiveDict(kwargs.get("headers"))  # type: Mapping[str, Any]
         error_map = {401: ClientAuthenticationError, 412: ResourceExistsError}
         try:
             key_value_added = self._impl.put_key_value(
                 entity=key_value,
-                key=key_value.key,
+                key=key_value.key,  # type: ignore
                 label=key_value.label,
                 if_none_match="*",
                 headers=custom_headers,
@@ -366,7 +364,7 @@ def set_configuration_setting(
             returned_config_setting = client.set_configuration_setting(config_setting)
         """
         key_value = configuration_setting._to_generated()
-        custom_headers = CaseInsensitiveDict(kwargs.get("headers"))
+        custom_headers = CaseInsensitiveDict(kwargs.get("headers"))  # type: Mapping[str, Any]
         error_map = {401: ClientAuthenticationError, 409: ResourceReadOnlyError}
         if match_condition == MatchConditions.IfNotModified:
             error_map[412] = ResourceModifiedError
@@ -380,7 +378,7 @@ def set_configuration_setting(
         try:
             key_value_set = self._impl.put_key_value(
                 entity=key_value,
-                key=key_value.key,
+                key=key_value.key,  # type: ignore
                 label=key_value.label,
                 if_match=prep_if_match(configuration_setting.etag, match_condition),
                 if_none_match=prep_if_none_match(
@@ -426,7 +424,7 @@ def delete_configuration_setting(self, key, label=None, **kwargs):
         """
         etag = kwargs.pop("etag", None)
         match_condition = kwargs.pop("match_condition", MatchConditions.Unconditionally)
-        custom_headers = CaseInsensitiveDict(kwargs.get("headers"))
+        custom_headers = CaseInsensitiveDict(kwargs.get("headers"))  # type: Mapping[str, Any]
         error_map = {401: ClientAuthenticationError, 409: ResourceReadOnlyError}
         if match_condition == MatchConditions.IfNotModified:
             error_map[412] = ResourceModifiedError
@@ -445,7 +443,7 @@ def delete_configuration_setting(self, key, label=None, **kwargs):
                 headers=custom_headers,
                 error_map=error_map,
             )
-            return ConfigurationSetting._from_generated(key_value_deleted)
+            return ConfigurationSetting._from_generated(key_value_deleted)  # type: ignore
         except HttpResponseError as error:
             e = error_map[error.status_code]
             raise e(message=error.message, response=error.response)
@@ -496,7 +494,7 @@ def list_revisions(self, key_filter=None, label_filter=None, **kwargs):
         error_map = {401: ClientAuthenticationError}
 
         try:
-            return self._impl.get_revisions(
+            return self._impl.get_revisions(  # type: ignore
                 label=label_filter,
                 key=key_filter,
                 select=select,

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

@@ -8,6 +8,11 @@
 from ._generated.models import KeyValue
 
 
+PolymorphicConfigurationSetting = Union[
+    "ConfigurationSetting", "SecretReferenceConfigurationSetting", "FeatureFlagConfigurationSetting"
+]
+
+
 class ConfigurationSetting(Model):
     """A configuration value.
     Variables are only populated by the server, and will be ignored when
@@ -59,14 +64,14 @@ def __init__(self, **kwargs):
 
     @classmethod
     def _from_generated(cls, key_value):
-        # type: (KeyValue) -> ConfigurationSetting
+        # type: (KeyValue) -> PolymorphicConfigurationSetting
         if key_value is None:
-            return None
+            return key_value
         if key_value.content_type is not None:
             try:
                 if key_value.content_type.startswith(
                     FeatureFlagConfigurationSetting._feature_flag_content_type  # pylint:disable=protected-access
-                ) and key_value.key.startswith(FeatureFlagConfigurationSetting.key_prefix):
+                ) and key_value.key.startswith(FeatureFlagConfigurationSetting.key_prefix):  # type: ignore
                     return FeatureFlagConfigurationSetting._from_generated(  # pylint: disable=protected-access
                         key_value
                     )
@@ -91,7 +96,7 @@ def _from_generated(cls, key_value):
         )
 
     def _to_generated(self):
-        # type: (...) -> KeyValue
+        # type: () -> KeyValue
         return KeyValue(
             key=self.key,
             label=self.label,
@@ -182,7 +187,7 @@ def enabled(self):
 
     @enabled.setter
     def enabled(self, new_value):
-        # type: (bool) -> bool
+        # type: (bool) -> None
         self._validate()
         if self.value is None:
             self.value = {}
@@ -215,28 +220,28 @@ def filters(self, new_filters):
 
     @classmethod
     def _from_generated(cls, key_value):
-        # type: (KeyValue) -> FeatureFlagConfigurationSetting
+        # type: (KeyValue) -> Union[FeatureFlagConfigurationSetting, ConfigurationSetting]
         try:
             if key_value is None:
-                return None
+                return key_value
             if key_value.value:
                 try:
                     key_value.value = json.loads(key_value.value)
                 except json.decoder.JSONDecodeError:
                     pass
 
-            filters = key_value.value["conditions"]["client_filters"]
+            filters = key_value.value["conditions"]["client_filters"]  # type: ignore
 
             return cls(
-                feature_id=key_value.key,
-                enabled=key_value.value["enabled"],
+                feature_id=key_value.key,  # type: ignore
+                enabled=key_value.value["enabled"],  # type: ignore
                 label=key_value.label,
                 content_type=key_value.content_type,
                 last_modified=key_value.last_modified,
                 tags=key_value.tags,
                 read_only=key_value.locked,
                 etag=key_value.etag,
-                filters=filters,
+                filters=filters,  # type: ignore
                 value=key_value.value,
             )
         except (KeyError, AttributeError):
@@ -273,13 +278,13 @@ class SecretReferenceConfigurationSetting(ConfigurationSetting):
     :param content_type:
     :type content_type: str
     :ivar value: The value of the configuration setting
-    :vartype value: str
+    :vartype value: Dict[str, Any]
     :ivar last_modified:
     :vartype last_modified: datetime
     :ivar read_only:
     :vartype read_only: bool
     :param tags:
-    :type tags: dict[str, str]
+    :type tags: Dict[str, str]
     """
 
     _attribute_map = {
@@ -336,16 +341,16 @@ def _validate(self):
     def _from_generated(cls, key_value):
         # type: (KeyValue) -> SecretReferenceConfigurationSetting
         if key_value is None:
-            return None
+            return key_value
         if key_value.value:
             try:
                 key_value.value = json.loads(key_value.value)
             except json.decoder.JSONDecodeError:
                 pass
 
         return cls(
-            key=key_value.key,
-            secret_uri=key_value.value[u"secret_uri"],
+            key=key_value.key,  # type: ignore
+            secret_uri=key_value.value[u"secret_uri"],  # type: ignore
             label=key_value.label,
             secret_id=key_value.value,
             last_modified=key_value.last_modified,
@@ -355,7 +360,7 @@ def _from_generated(cls, key_value):
         )
 
     def _to_generated(self):
-        # type: (...) -> KeyValue
+        # type: () -> KeyValue
         return KeyValue(
             key=self.key,
             label=self.label,

sdk/appconfiguration/azure-appconfiguration/azure/appconfiguration/_sync_token.py

@@ -23,6 +23,7 @@
 # IN THE SOFTWARE.
 #
 # --------------------------------------------------------------------------
+from typing import Any, Dict
 from azure.core.pipeline import PipelineRequest, PipelineResponse
 from azure.core.pipeline.policies import SansIOHTTPPolicy
 
@@ -61,7 +62,7 @@ class SyncTokenPolicy(SansIOHTTPPolicy):
     def __init__(self, **kwargs):  # pylint: disable=unused-argument
         # type: (**Any) -> None
         self._sync_token_header = "Sync-Token"
-        self._sync_tokens = dict()
+        self._sync_tokens = {}  # type: Dict[str, Any]
 
     def on_request(self, request):  # type: ignore # pylint: disable=arguments-differ
         # type: (PipelineRequest) -> None

sdk/appconfiguration/azure-appconfiguration/azure/appconfiguration/_utils.py

@@ -5,10 +5,12 @@
 # -------------------------------------------------------------------------
 
 from datetime import datetime
+from typing import Optional, Tuple
 from azure.core import MatchConditions
 
 
 def quote_etag(etag):
+    # type: (Optional[str]) -> Optional[str]
     if not etag or etag == "*":
         return etag
     if etag.startswith('"') and etag.endswith('"'):
@@ -19,7 +21,7 @@ def quote_etag(etag):
 
 
 def prep_if_match(etag, match_condition):
-    # type: (str, MatchConditions) -> str
+    # type: (Optional[str], Optional[MatchConditions]) -> Optional[str]
     if match_condition == MatchConditions.IfNotModified:
         if_match = quote_etag(etag) if etag else None
         return if_match
@@ -29,7 +31,7 @@ def prep_if_match(etag, match_condition):
 
 
 def prep_if_none_match(etag, match_condition):
-    # type: (str, MatchConditions) -> str
+    # type: (Optional[str], Optional[MatchConditions]) -> Optional[str]
     if match_condition == MatchConditions.IfModified:
         if_none_match = quote_etag(etag) if etag else None
         return if_none_match
@@ -39,11 +41,13 @@ def prep_if_none_match(etag, match_condition):
 
 
 def get_endpoint_from_connection_string(connection_string):
+    # type: (str) -> str
     endpoint, _, _ = parse_connection_string(connection_string)
     return endpoint
 
 
 def parse_connection_string(connection_string):
+    # type: (str) -> Tuple[str, str, str]
     # connection_string looks like Endpoint=https://xxxxx;Id=xxxxx;Secret=xxxx
     segments = connection_string.split(";")
     if len(segments) != 3:
@@ -70,4 +74,5 @@ def parse_connection_string(connection_string):
 
 
 def get_current_utc_time():
+    # type: () -> str
     return str(datetime.utcnow().strftime("%b, %d %Y %H:%M:%S.%f ")) + "GMT"