[ACR] Update method type hints to Python3 syntax (Azure#28117)

Author: Yalin Li

sdk/containerregistry/azure-containerregistry/azure/containerregistry/_anonymous_exchange_client.py

@@ -3,7 +3,7 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 # ------------------------------------
-from typing import Any, Union, Optional
+from typing import Optional
 
 from ._exchange_client import ExchangeClientAuthenticationPolicy
 from ._generated import ContainerRegistry
@@ -18,12 +18,11 @@ class AnonymousACRExchangeClient(object):
     :param endpoint: Azure Container Registry endpoint
     :type endpoint: str
     :keyword api_version: API Version. The default value is "2021-07-01". Note that overriding this default value
-     may result in unsupported behavior.
+        may result in unsupported behavior.
     :paramtype api_version: str
     """
 
-    def __init__(self, endpoint, **kwargs):  # pylint: disable=missing-client-constructor-parameter-credential
-        # type: (str, Any) -> None
+    def __init__(self, endpoint: str, **kwargs) -> None: # pylint: disable=missing-client-constructor-parameter-credential
         if not endpoint.startswith("https://") and not endpoint.startswith("http://"):
             endpoint = "https://" + endpoint
         self._endpoint = endpoint
@@ -35,8 +34,7 @@ def __init__(self, endpoint, **kwargs):  # pylint: disable=missing-client-constr
             **kwargs
         )
 
-    def get_acr_access_token(self, challenge, **kwargs):
-        # type: (str, Any) -> Optional[str]
+    def get_acr_access_token(self, challenge: str, **kwargs) -> Optional[str]:
         parsed_challenge = _parse_challenge(challenge)
         return self.exchange_refresh_token_for_access_token(
             "",
@@ -46,8 +44,9 @@ def get_acr_access_token(self, challenge, **kwargs):
             **kwargs
         )
 
-    def exchange_refresh_token_for_access_token(self, refresh_token, service, scope, grant_type, **kwargs):
-        # type: (str, str, str, Union[str, TokenGrantType], Any) -> Optional[str]
+    def exchange_refresh_token_for_access_token(
+        self, refresh_token: str, service: str, scope: str, grant_type: str, **kwargs
+    ) -> Optional[str]:
         access_token = self._client.authentication.exchange_acr_refresh_token_for_acr_access_token(
             service=service, scope=scope, refresh_token=refresh_token, grant_type=grant_type, **kwargs
         )
@@ -60,8 +59,7 @@ def __enter__(self):
     def __exit__(self, *args):
         self._client.__exit__(*args)
 
-    def close(self):
-        # type: () -> None
+    def close(self) -> None:
         """Close sockets opened by the client.
         Calling this method is unnecessary when using the client as a context manager.
         """

sdk/containerregistry/azure-containerregistry/azure/containerregistry/_authentication_policy.py

@@ -4,44 +4,41 @@
 # Licensed under the MIT License.
 # ------------------------------------
 
-from typing import TYPE_CHECKING, Any, Union, Optional
+from typing import Union, Optional
 from io import SEEK_SET, UnsupportedOperation
 
+from azure.core.credentials import TokenCredential
+from azure.core.pipeline import PipelineRequest, PipelineResponse
 from azure.core.pipeline.policies import HTTPPolicy
 
 from ._anonymous_exchange_client import AnonymousACRExchangeClient
 from ._exchange_client import ACRExchangeClient
 from ._helpers import _enforce_https
 
-if TYPE_CHECKING:
-    from azure.core.credentials import TokenCredential
-    from azure.core.pipeline import PipelineRequest, PipelineResponse
-
 
 class ContainerRegistryChallengePolicy(HTTPPolicy):
     """Authentication policy for ACR which accepts a challenge"""
 
-    def __init__(self, credential, endpoint, **kwargs):
-        # type: (Optional[TokenCredential], str, **Any) -> None
+    def __init__(self, credential: Optional[TokenCredential], endpoint: str, **kwargs) -> None:
         super(ContainerRegistryChallengePolicy, self).__init__()
         self._credential = credential
         if self._credential is None:
             self._exchange_client = AnonymousACRExchangeClient(endpoint, **kwargs) # type: Union[AnonymousACRExchangeClient, ACRExchangeClient] # pylint: disable=line-too-long
         else:
             self._exchange_client = ACRExchangeClient(endpoint, self._credential, **kwargs)
 
-    def on_request(self, request):
-        # type: (PipelineRequest) -> None
+    def on_request(self, request: PipelineRequest) -> None:
         """Called before the policy sends a request.
         The base implementation authorizes the request with a bearer token.
+
         :param ~azure.core.pipeline.PipelineRequest request: the request
         """
         # Future caching implementation will be included here
         pass  # pylint: disable=unnecessary-pass
 
-    def send(self, request):
-        # type: (PipelineRequest) -> PipelineResponse
+    def send(self, request: PipelineRequest) -> PipelineResponse:
         """Authorizes a request with a bearer token, possibly handling an authentication challenge
+
         :param ~azure.core.pipeline.PipelineRequest request: the request
         """
         _enforce_https(request)
@@ -65,16 +62,16 @@ def send(self, request):
 
         return response
 
-    def on_challenge(self, request, response, challenge):
-        # type: (PipelineRequest, PipelineResponse, str) -> bool
+    def on_challenge(self, request: PipelineRequest, response: PipelineResponse, challenge: str) -> bool:
         """Authorize request according to an authentication challenge
         This method is called when the resource provider responds 401 with a WWW-Authenticate header.
+
         :param ~azure.core.pipeline.PipelineRequest request: the request which elicited an authentication challenge
         :param ~azure.core.pipeline.PipelineResponse response: the resource provider's response
         :param str challenge: response's WWW-Authenticate header, unparsed. It may contain multiple challenges.
         :returns: a bool indicating whether the policy should send the request
         """
-        # pylint:disable=unused-argument,no-self-use
+        # pylint:disable=unused-argument, no-self-use
 
         access_token = self._exchange_client.get_acr_access_token(challenge)
         if access_token is not None:

sdk/containerregistry/azure-containerregistry/azure/containerregistry/_base_client.py

@@ -4,19 +4,16 @@
 # Licensed under the MIT License.
 # ------------------------------------
 from enum import Enum
-from typing import TYPE_CHECKING, Any, Optional
+from typing import Optional
 
 from azure.core import CaseInsensitiveEnumMeta
+from azure.core.credentials import TokenCredential
 from azure.core.pipeline.transport import HttpTransport
 
 from ._authentication_policy import ContainerRegistryChallengePolicy
 from ._generated import ContainerRegistry
 from ._user_agent import USER_AGENT
 
-if TYPE_CHECKING:
-    from azure.core.credentials import TokenCredential
-
-
 
 class ContainerRegistryApiVersion(str, Enum, metaclass=CaseInsensitiveEnumMeta):
     """Container Registry API version supported by this package"""
@@ -33,12 +30,11 @@ class ContainerRegistryBaseClient(object):
     :keyword credential_scopes: URL for credential authentication if different from the default
     :paramtype credential_scopes: List[str]
     :keyword api_version: API Version. The default value is "2021-07-01". Note that overriding this default value
-     may result in unsupported behavior.
+        may result in unsupported behavior.
     :paramtype api_version: str
     """
 
-    def __init__(self, endpoint, credential, **kwargs):
-        # type: (str, Optional[TokenCredential], Any) -> None
+    def __init__(self, endpoint: str, credential: Optional[TokenCredential], **kwargs) -> None:
         self._auth_policy = ContainerRegistryChallengePolicy(credential, endpoint, **kwargs)
         self._client = ContainerRegistry(
             credential=credential,
@@ -57,8 +53,7 @@ def __exit__(self, *args):
         self._auth_policy.__exit__(*args)
         self._client.__exit__(*args)
 
-    def close(self):
-        # type: () -> None
+    def close(self) -> None:
         """Close sockets opened by the client.
         Calling this method is unnecessary when using the client as a context manager.
         """