[Key Vault] Resolve next-pylint and next-mypy errors (#33448)

Author: mccoyp

sdk/keyvault/azure-keyvault-administration/README.md

@@ -278,7 +278,7 @@ to the library's [credential documentation][sas_docs]. Alternatively, it is poss
 CONTAINER_URL = os.environ["CONTAINER_URL"]
 SAS_TOKEN = os.environ["SAS_TOKEN"]
 
-backup_result: KeyVaultBackupResult = client.begin_backup(CONTAINER_URL, SAS_TOKEN).result()
+backup_result: KeyVaultBackupResult = client.begin_backup(CONTAINER_URL, sas_token=SAS_TOKEN).result()
 print(f"Azure Storage Blob URL of the backup: {backup_result.folder_url}")
 ```
 
@@ -297,14 +297,14 @@ For more details on creating a SAS token using a `BlobServiceClient` from [`azur
 to the library's [credential documentation][sas_docs]. Alternatively, it is possible to
 [generate a SAS token in Storage Explorer][storage_explorer].
 
-<!-- SNIPPET:backup_restore_operations.begin_backup -->
+<!-- SNIPPET:backup_restore_operations.begin_restore -->
 
 ```python
-CONTAINER_URL = os.environ["CONTAINER_URL"]
 SAS_TOKEN = os.environ["SAS_TOKEN"]
 
-backup_result: KeyVaultBackupResult = client.begin_backup(CONTAINER_URL, SAS_TOKEN).result()
-print(f"Azure Storage Blob URL of the backup: {backup_result.folder_url}")
+# `backup_result` is the KeyVaultBackupResult returned by `begin_backup`
+client.begin_restore(backup_result.folder_url, sas_token=SAS_TOKEN).wait()
+print("Vault restored successfully.")
 ```
 
 <!-- END SNIPPET -->

sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_access_control_client.py

@@ -2,22 +2,17 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 # ------------------------------------
-from typing import TYPE_CHECKING
-from uuid import uuid4
+from typing import Any, Union
+from uuid import UUID, uuid4
 
 from azure.core.exceptions import ResourceNotFoundError
+from azure.core.paging import ItemPaged
 from azure.core.tracing.decorator import distributed_trace
 
+from ._enums import KeyVaultRoleScope
 from ._models import KeyVaultRoleAssignment, KeyVaultRoleDefinition
 from ._internal import KeyVaultClientBase
 
-if TYPE_CHECKING:
-    # pylint:disable=ungrouped-imports
-    from typing import Union
-    from uuid import UUID
-    from azure.core.paging import ItemPaged
-    from ._enums import KeyVaultRoleScope
-
 
 class KeyVaultAccessControlClient(KeyVaultClientBase):
     """Manages role-based access to Azure Key Vault.
@@ -27,7 +22,7 @@ class KeyVaultAccessControlClient(KeyVaultClientBase):
         See https://aka.ms/azsdk/blog/vault-uri for details.
     :param credential: An object which can provide an access token for the vault, such as a credential from
         :mod:`azure.identity`
-    :type credential: :class:`~azure.core.credentials.TokenCredential`
+    :type credential: ~azure.core.credentials.TokenCredential
 
     :keyword api_version: Version of the service API to use. Defaults to the most recent.
     :paramtype api_version: ~azure.keyvault.administration.ApiVersion or str
@@ -39,7 +34,7 @@ class KeyVaultAccessControlClient(KeyVaultClientBase):
 
     @distributed_trace
     def create_role_assignment(
-        self, scope: "Union[str, KeyVaultRoleScope]", definition_id: str, principal_id: str, **kwargs
+        self, scope: Union[str, KeyVaultRoleScope], definition_id: str, principal_id: str, **kwargs: Any
     ) -> KeyVaultRoleAssignment:
         """Create a role assignment.
 
@@ -74,7 +69,7 @@ def create_role_assignment(
 
     @distributed_trace
     def delete_role_assignment(
-        self, scope: "Union[str, KeyVaultRoleScope]", name: "Union[str, UUID]", **kwargs
+        self, scope: Union[str, KeyVaultRoleScope], name: Union[str, UUID], **kwargs: Any
     ) -> None:
         """Delete a role assignment.
 
@@ -96,7 +91,7 @@ def delete_role_assignment(
 
     @distributed_trace
     def get_role_assignment(
-        self, scope: "Union[str, KeyVaultRoleScope]", name: "Union[str, UUID]", **kwargs
+        self, scope: Union[str, KeyVaultRoleScope], name: Union[str, UUID], **kwargs: Any
     ) -> KeyVaultRoleAssignment:
         """Get a role assignment.
 
@@ -116,8 +111,8 @@ def get_role_assignment(
 
     @distributed_trace
     def list_role_assignments(
-        self, scope: "Union[str, KeyVaultRoleScope]", **kwargs
-    ) -> "ItemPaged[KeyVaultRoleAssignment]":
+        self, scope: Union[str, KeyVaultRoleScope], **kwargs: Any
+    ) -> ItemPaged[KeyVaultRoleAssignment]:
         """List all role assignments for a scope.
 
         :param scope: scope of the role assignments. :class:`KeyVaultRoleScope` defines common broad scopes.
@@ -136,8 +131,8 @@ def list_role_assignments(
 
     @distributed_trace
     def set_role_definition(
-        self, scope: "Union[str, KeyVaultRoleScope]", **kwargs
-    ) -> "KeyVaultRoleDefinition":
+        self, scope: Union[str, KeyVaultRoleScope], **kwargs: Any
+    ) -> KeyVaultRoleDefinition:
         """Creates or updates a custom role definition.
 
         To update a role definition, specify the definition's ``name``.
@@ -194,8 +189,8 @@ def set_role_definition(
 
     @distributed_trace
     def get_role_definition(
-        self, scope: "Union[str, KeyVaultRoleScope]", name: "Union[str, UUID]", **kwargs
-    ) -> "KeyVaultRoleDefinition":
+        self, scope: Union[str, KeyVaultRoleScope], name: Union[str, UUID], **kwargs: Any
+    ) -> KeyVaultRoleDefinition:
         """Get the specified role definition.
 
         :param scope: scope of the role definition. :class:`KeyVaultRoleScope` defines common broad scopes.
@@ -214,7 +209,7 @@ def get_role_definition(
 
     @distributed_trace
     def delete_role_definition(
-        self, scope: "Union[str, KeyVaultRoleScope]", name: "Union[str, UUID]", **kwargs
+        self, scope: Union[str, KeyVaultRoleScope], name: Union[str, UUID], **kwargs: Any
     ) -> None:
         """Deletes a custom role definition.
 
@@ -236,8 +231,8 @@ def delete_role_definition(
 
     @distributed_trace
     def list_role_definitions(
-        self, scope: "Union[str, KeyVaultRoleScope]", **kwargs
-    ) -> "ItemPaged[KeyVaultRoleDefinition]":
+        self, scope: Union[str, KeyVaultRoleScope], **kwargs: Any
+    ) -> ItemPaged[KeyVaultRoleDefinition]:
         """List all role definitions applicable at and above a scope.
 
         :param scope: scope of the role definitions. :class:`KeyVaultRoleScope` defines common broad scopes.

sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_backup_client.py

@@ -5,21 +5,18 @@
 import base64
 import functools
 import pickle
-from typing import Any, Optional, overload, TYPE_CHECKING
+from typing import Any, Optional, overload
 from urllib.parse import urlparse
 
 from typing_extensions import Literal
 
+from azure.core.polling import LROPoller
 from azure.core.tracing.decorator import distributed_trace
 
 from ._models import KeyVaultBackupResult
 from ._internal import KeyVaultClientBase, parse_folder_url
 from ._internal.polling import KeyVaultBackupClientPolling, KeyVaultBackupClientPollingMethod
 
-if TYPE_CHECKING:
-    # pylint:disable=unused-import
-    from azure.core.polling import LROPoller
-
 
 def _parse_status_url(url):
     parsed = urlparse(url)
@@ -35,7 +32,7 @@ class KeyVaultBackupClient(KeyVaultClientBase):
         See https://aka.ms/azsdk/blog/vault-uri for details.
     :param credential: An object which can provide an access token for the vault, such as a credential from
         :mod:`azure.identity`
-    :type credential: :class:`~azure.core.credentials.TokenCredential`
+    :type credential: ~azure.core.credentials.TokenCredential
 
     :keyword api_version: Version of the service API to use. Defaults to the most recent.
     :paramtype api_version: ~azure.keyvault.administration.ApiVersion or str
@@ -51,7 +48,7 @@ def begin_backup(
         use_managed_identity: Literal[True],
         continuation_token: Optional[str] = None,
         **kwargs: Any,
-    ) -> "LROPoller[KeyVaultBackupResult]":
+    ) -> LROPoller[KeyVaultBackupResult]:
         ...
 
     @overload
@@ -62,11 +59,11 @@ def begin_backup(
         sas_token: str,
         continuation_token: Optional[str] = None,
         **kwargs: Any,
-    ) -> "LROPoller[KeyVaultBackupResult]":
+    ) -> LROPoller[KeyVaultBackupResult]:
         ...
 
     @distributed_trace
-    def begin_backup(self, blob_storage_url: str, *args: str, **kwargs: Any) -> "LROPoller[KeyVaultBackupResult]":
+    def begin_backup(self, blob_storage_url: str, *args: str, **kwargs: Any) -> LROPoller[KeyVaultBackupResult]:
         """Begin a full backup of the Key Vault.
 
         :param str blob_storage_url: URL of the blob storage container in which the backup will be stored, for example
@@ -142,7 +139,7 @@ def begin_restore(
         key_name: Optional[str] = None,
         continuation_token: Optional[str] = None,
         **kwargs: Any,
-    ) -> "LROPoller":
+    ) -> LROPoller[None]:
         ...
 
     @overload
@@ -154,11 +151,11 @@ def begin_restore(
         key_name: Optional[str] = None,
         continuation_token: Optional[str] = None,
         **kwargs: Any,
-    ) -> "LROPoller":
+    ) -> LROPoller[None]:
         ...
 
     @distributed_trace
-    def begin_restore(self, folder_url: str, *args: str, **kwargs: Any) -> "LROPoller":
+    def begin_restore(self, folder_url: str, *args: str, **kwargs: Any) -> LROPoller[None]:
         """Restore a Key Vault backup.
 
         This method restores either a complete Key Vault backup or when ``key_name`` has a value, a single key.

sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_internal/async_challenge_auth_policy.py

@@ -15,36 +15,33 @@
 """
 
 import time
-from typing import TYPE_CHECKING
+from typing import Any, Optional
 from urllib.parse import urlparse
 
+from azure.core.credentials import AccessToken
+from azure.core.credentials_async import AsyncTokenCredential
+from azure.core.pipeline import PipelineRequest, PipelineResponse
 from azure.core.pipeline.policies import AsyncBearerTokenCredentialPolicy
 
 from . import http_challenge_cache as ChallengeCache
 from .challenge_auth_policy import _enforce_tls, _update_challenge
 
-if TYPE_CHECKING:
-    from typing import Optional
-    from azure.core.credentials import AccessToken
-    from azure.core.credentials_async import AsyncTokenCredential
-    from azure.core.pipeline import PipelineRequest, PipelineResponse
-
 
 class AsyncChallengeAuthPolicy(AsyncBearerTokenCredentialPolicy):
     """Policy for handling HTTP authentication challenges.
 
     :param credential: An object which can provide an access token for the vault, such as a credential from
         :mod:`azure.identity.aio`
-    :type credential: :class:`~azure.core.credentials_async.AsyncTokenCredential`
+    :type credential: ~azure.core.credentials_async.AsyncTokenCredential
     """
 
-    def __init__(self, credential: "AsyncTokenCredential", *scopes: str, **kwargs) -> None:
+    def __init__(self, credential: AsyncTokenCredential, *scopes: str, **kwargs: Any) -> None:
         super().__init__(credential, *scopes, **kwargs)
         self._credential = credential
-        self._token: "Optional[AccessToken]" = None
+        self._token: Optional[AccessToken] = None
         self._verify_challenge_resource = kwargs.pop("verify_challenge_resource", True)
 
-    async def on_request(self, request: "PipelineRequest") -> None:
+    async def on_request(self, request: PipelineRequest) -> None:
         _enforce_tls(request)
         challenge = ChallengeCache.get_challenge_for_url(request.http_request.url)
         if challenge:
@@ -68,7 +65,7 @@ async def on_request(self, request: "PipelineRequest") -> None:
             request.http_request.headers["Content-Length"] = "0"
 
 
-    async def on_challenge(self, request: "PipelineRequest", response: "PipelineResponse") -> bool:
+    async def on_challenge(self, request: PipelineRequest, response: PipelineResponse) -> bool:
         try:
             challenge = _update_challenge(request, response)
             # azure-identity credentials require an AADv2 scope but the challenge may specify an AADv1 resource

sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_internal/async_client_base.py

@@ -2,9 +2,11 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 # ------------------------------------
-from typing import TYPE_CHECKING
+from typing import Any, Awaitable
 
+from azure.core.credentials_async import AsyncTokenCredential
 from azure.core.pipeline.policies import HttpLoggingPolicy
+from azure.core.rest import AsyncHttpResponse, HttpRequest
 from azure.core.tracing.decorator_async import distributed_trace_async
 
 from . import AsyncChallengeAuthPolicy
@@ -13,16 +15,10 @@
 from .._generated.aio import KeyVaultClient as _KeyVaultClient
 from .._generated import models as _models
 
-if TYPE_CHECKING:
-    # pylint:disable=unused-import
-    from typing import Any, Awaitable
-    from azure.core.credentials_async import AsyncTokenCredential
-    from azure.core.rest import AsyncHttpResponse, HttpRequest
-
 
 class AsyncKeyVaultClientBase(object):
     # pylint:disable=protected-access
-    def __init__(self, vault_url: str, credential: "AsyncTokenCredential", **kwargs) -> None:
+    def __init__(self, vault_url: str, credential: AsyncTokenCredential, **kwargs: Any) -> None:
         if not credential:
             raise ValueError(
                 "credential should be an object supporting the AsyncTokenCredential protocol, "
@@ -88,7 +84,9 @@ async def close(self) -> None:
         await self._client.close()
 
     @distributed_trace_async
-    def send_request(self, request: "HttpRequest", *, stream: bool = False, **kwargs) -> "Awaitable[AsyncHttpResponse]":
+    def send_request(
+        self, request: HttpRequest, *, stream: bool = False, **kwargs: Any
+    ) -> Awaitable[AsyncHttpResponse]:
         """Runs a network request using the client's existing pipeline.
 
         The request URL can be relative to the vault URL. The service API version used for the request is the same as

sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_internal/challenge_auth_policy.py

@@ -15,21 +15,17 @@
 """
 
 import time
-from typing import TYPE_CHECKING
+from typing import Any, Optional
 from urllib.parse import urlparse
 
+from azure.core.credentials import AccessToken, TokenCredential
 from azure.core.exceptions import ServiceRequestError
-from azure.core.pipeline import PipelineRequest
+from azure.core.pipeline import PipelineRequest, PipelineResponse
 from azure.core.pipeline.policies import BearerTokenCredentialPolicy
 
 from .http_challenge import HttpChallenge
 from . import http_challenge_cache as ChallengeCache
 
-if TYPE_CHECKING:
-    from typing import Optional
-    from azure.core.credentials import AccessToken, TokenCredential
-    from azure.core.pipeline import PipelineResponse
-
 
 def _enforce_tls(request: PipelineRequest) -> None:
     if not request.http_request.url.lower().startswith("https"):
@@ -38,13 +34,13 @@ def _enforce_tls(request: PipelineRequest) -> None:
         )
 
 
-def _update_challenge(request: PipelineRequest, challenger: "PipelineResponse") -> HttpChallenge:
+def _update_challenge(request: PipelineRequest, challenger: PipelineResponse) -> HttpChallenge:
     """Parse challenge from a challenge response, cache it, and return it.
 
     :param request: The pipeline request that prompted the challenge response.
-    :type request: :class:`~azure.core.pipeline.PipelineRequest`
+    :type request: ~azure.core.pipeline.PipelineRequest
     :param challenger: The pipeline response containing the authentication challenge.
-    :type challenger: :class:`~azure.core.pipeline.PipelineResponse`
+    :type challenger: ~azure.core.pipeline.PipelineResponse
 
     :returns: An HttpChallenge object representing the authentication challenge.
     :rtype: HttpChallenge
@@ -64,10 +60,10 @@ class ChallengeAuthPolicy(BearerTokenCredentialPolicy):
 
     :param credential: An object which can provide an access token for the vault, such as a credential from
         :mod:`azure.identity`
-    :type credential: :class:`~azure.core.credentials.TokenCredential`
+    :type credential: ~azure.core.credentials.TokenCredential
     """
 
-    def __init__(self, credential: "TokenCredential", *scopes: str, **kwargs) -> None:
+    def __init__(self, credential: TokenCredential, *scopes: str, **kwargs: Any) -> None:
         super(ChallengeAuthPolicy, self).__init__(credential, *scopes, **kwargs)
         self._credential = credential
         self._token: "Optional[AccessToken]" = None
@@ -96,7 +92,7 @@ def on_request(self, request: PipelineRequest) -> None:
             request.http_request.set_json_body(None)
             request.http_request.headers["Content-Length"] = "0"
 
-    def on_challenge(self, request: PipelineRequest, response: "PipelineResponse") -> bool:
+    def on_challenge(self, request: PipelineRequest, response: PipelineResponse) -> bool:
         try:
             challenge = _update_challenge(request, response)
             # azure-identity credentials require an AADv2 scope but the challenge may specify an AADv1 resource