[Key Vault] Python 3-style typing and formatting (Azure#28113)

Author: mccoyp

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

@@ -38,8 +38,9 @@ class KeyVaultAccessControlClient(KeyVaultClientBase):
     # pylint:disable=protected-access
 
     @distributed_trace
-    def create_role_assignment(self, scope, definition_id, principal_id, **kwargs):
-        # type: (Union[str, KeyVaultRoleScope], str, str, **Any) -> KeyVaultRoleAssignment
+    def create_role_assignment(
+        self, scope: "Union[str, KeyVaultRoleScope]", definition_id: str, principal_id: str, **kwargs
+    ) -> KeyVaultRoleAssignment:
         """Create a role assignment.
 
         :param scope: scope the role assignment will apply over. :class:`KeyVaultRoleScope` defines common
@@ -48,8 +49,10 @@ def create_role_assignment(self, scope, definition_id, principal_id, **kwargs):
         :param str definition_id: ID of the role's definition
         :param str principal_id: Azure Active Directory object ID of the principal which will be assigned the role. The
             principal can be a user, service principal, or security group.
+
         :keyword name: a name for the role assignment. Must be a UUID.
         :paramtype name: str or uuid.UUID
+
         :rtype: ~azure.keyvault.administration.KeyVaultRoleAssignment
         """
         name = kwargs.pop("name", None) or uuid4()
@@ -69,15 +72,17 @@ def create_role_assignment(self, scope, definition_id, principal_id, **kwargs):
         return KeyVaultRoleAssignment._from_generated(assignment)
 
     @distributed_trace
-    def delete_role_assignment(self, scope, name, **kwargs):
-        # type: (Union[str, KeyVaultRoleScope], Union[str, UUID], **Any) -> None
+    def delete_role_assignment(
+        self, scope: "Union[str, KeyVaultRoleScope]", name: "Union[str, UUID]", **kwargs
+    ) -> None:
         """Delete a role assignment.
 
         :param scope: the assignment's scope, for example "/", "/keys", or "/keys/<specific key identifier>"
             :class:`KeyVaultRoleScope` defines common broad scopes. Specify a narrower scope as a string.
         :type scope: str or KeyVaultRoleScope
         :param name: the role assignment's name.
         :type name: str or uuid.UUID
+
         :returns: None
         """
         try:
@@ -88,15 +93,17 @@ def delete_role_assignment(self, scope, name, **kwargs):
             pass
 
     @distributed_trace
-    def get_role_assignment(self, scope, name, **kwargs):
-        # type: (Union[str, KeyVaultRoleScope], Union[str, UUID], **Any) -> KeyVaultRoleAssignment
+    def get_role_assignment(
+        self, scope: "Union[str, KeyVaultRoleScope]", name: "Union[str, UUID]", **kwargs
+    ) -> KeyVaultRoleAssignment:
         """Get a role assignment.
 
         :param scope: the assignment's scope, for example "/", "/keys", or "/keys/<specific key identifier>"
             :class:`KeyVaultRoleScope` defines common broad scopes. Specify a narrower scope as a string.
         :type scope: str or KeyVaultRoleScope
         :param name: the role assignment's name.
         :type name: str or uuid.UUID
+
         :rtype: ~azure.keyvault.administration.KeyVaultRoleAssignment
         """
         assignment = self._client.role_assignments.get(
@@ -105,13 +112,15 @@ def get_role_assignment(self, scope, name, **kwargs):
         return KeyVaultRoleAssignment._from_generated(assignment)
 
     @distributed_trace
-    def list_role_assignments(self, scope, **kwargs):
-        # type: (Union[str, KeyVaultRoleScope], **Any) -> ItemPaged[KeyVaultRoleAssignment]
+    def list_role_assignments(
+        self, scope: "Union[str, KeyVaultRoleScope]", **kwargs
+    ) -> "ItemPaged[KeyVaultRoleAssignment]":
         """List all role assignments for a scope.
 
         :param scope: scope of the role assignments. :class:`KeyVaultRoleScope` defines common broad scopes.
             Specify a narrower scope as a string.
         :type scope: str or KeyVaultRoleScope
+
         :rtype: ~azure.core.paging.ItemPaged[~azure.keyvault.administration.KeyVaultRoleAssignment]
         """
         return self._client.role_assignments.list_for_scope(
@@ -122,15 +131,17 @@ def list_role_assignments(self, scope, **kwargs):
         )
 
     @distributed_trace
-    def set_role_definition(self, scope, **kwargs):
-        # type: (Union[str, KeyVaultRoleScope], **Any) -> KeyVaultRoleDefinition
+    def set_role_definition(
+        self, scope: "Union[str, KeyVaultRoleScope]", **kwargs
+    ) -> "KeyVaultRoleDefinition":
         """Creates or updates a custom role definition.
 
         To update a role definition, specify the definition's ``name``.
 
         :param scope: scope of the role definition. :class:`KeyVaultRoleScope` defines common broad scopes.
             Specify a narrower scope as a string. Managed HSM only supports '/', or KeyVaultRoleScope.GLOBAL.
         :type scope: str or KeyVaultRoleScope
+
         :keyword name: the role definition's name, a UUID. When this argument has a value, the client will create a new
             role definition with this name or update an existing role definition, if one exists with the given name.
             When this argument has no value, a new role definition will be created with a generated name.
@@ -144,6 +155,7 @@ def set_role_definition(self, scope, **kwargs):
         :paramtype permissions: Iterable[KeyVaultPermission]
         :keyword assignable_scopes: the scopes for which the role definition can be assigned.
         :paramtype assignable_scopes: Iterable[str] or Iterable[KeyVaultRoleScope]
+
         :returns: The created or updated role definition
         :rtype: ~azure.keyvault.administration.KeyVaultRoleDefinition
         """
@@ -175,15 +187,17 @@ def set_role_definition(self, scope, **kwargs):
         return KeyVaultRoleDefinition._from_generated(definition)
 
     @distributed_trace
-    def get_role_definition(self, scope, name, **kwargs):
-        # type: (Union[str, KeyVaultRoleScope], Union[str, UUID], **Any) -> KeyVaultRoleDefinition
+    def get_role_definition(
+        self, scope: "Union[str, KeyVaultRoleScope]", name: "Union[str, UUID]", **kwargs
+    ) -> "KeyVaultRoleDefinition":
         """Get the specified role definition.
 
         :param scope: scope of the role definition. :class:`KeyVaultRoleScope` defines common broad scopes.
             Specify a narrower scope as a string. Managed HSM only supports '/', or KeyVaultRoleScope.GLOBAL.
         :type scope: str or KeyVaultRoleScope
         :param name: the role definition's name.
         :type name: str or uuid.UUID
+
         :rtype: ~azure.keyvault.administration.KeyVaultRoleDefinition
         """
         definition = self._client.role_definitions.get(
@@ -192,15 +206,17 @@ def get_role_definition(self, scope, name, **kwargs):
         return KeyVaultRoleDefinition._from_generated(definition)
 
     @distributed_trace
-    def delete_role_definition(self, scope, name, **kwargs):
-        # type: (Union[str, KeyVaultRoleScope], Union[str, UUID], **Any) -> None
+    def delete_role_definition(
+        self, scope: "Union[str, KeyVaultRoleScope]", name: "Union[str, UUID]", **kwargs
+    ) -> None:
         """Deletes a custom role definition.
 
         :param scope: scope of the role definition. :class:`KeyVaultRoleScope` defines common broad scopes.
             Specify a narrower scope as a string. Managed HSM only supports '/', or KeyVaultRoleScope.GLOBAL.
         :type scope: str or KeyVaultRoleScope
         :param name: the role definition's name.
         :type name: str or uuid.UUID
+
         :returns: None
         """
         try:
@@ -211,13 +227,15 @@ def delete_role_definition(self, scope, name, **kwargs):
             pass
 
     @distributed_trace
-    def list_role_definitions(self, scope, **kwargs):
-        # type: (Union[str, KeyVaultRoleScope], **Any) -> ItemPaged[KeyVaultRoleDefinition]
+    def list_role_definitions(
+        self, scope: "Union[str, KeyVaultRoleScope]", **kwargs
+    ) -> "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.
             Specify a narrower scope as a string.
         :type scope: str or KeyVaultRoleScope
+
         :rtype: ~azure.core.paging.ItemPaged[~azure.keyvault.administration.KeyVaultRoleDefinition]
         """
         return self._client.role_definitions.list(

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

@@ -42,14 +42,17 @@ class KeyVaultBackupClient(KeyVaultClientBase):
     """
 
     # pylint:disable=protected-access
-    def begin_backup(self, blob_storage_url, sas_token, **kwargs):
-        # type: (str, str, **Any) -> LROPoller[KeyVaultBackupResult]
+    def begin_backup(
+        self, blob_storage_url: str, sas_token: str, **kwargs
+    ) -> "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
             https://<account>.blob.core.windows.net/backup
         :param str sas_token: a Shared Access Signature (SAS) token authorizing access to the blob storage resource
+
         :keyword str continuation_token: a continuation token to restart polling from a saved state
+
         :returns: An :class:`~azure.core.polling.LROPoller` instance. Call `result()` on this object to wait for the
             operation to complete and get a :class:`KeyVaultBackupResult`.
         :rtype: ~azure.core.polling.LROPoller[~azure.keyvault.administration.KeyVaultBackupResult]
@@ -95,8 +98,7 @@ def begin_backup(self, blob_storage_url, sas_token, **kwargs):
             **kwargs
         )
 
-    def begin_restore(self, folder_url, sas_token, **kwargs):
-        # type: (str, str, **Any) -> LROPoller
+    def begin_restore(self, folder_url: str, sas_token: str, **kwargs) -> "LROPoller":
         """Restore a Key Vault backup.
 
         This method restores either a complete Key Vault backup or when ``key_name`` has a value, a single key.
@@ -105,8 +107,10 @@ def begin_restore(self, folder_url, sas_token, **kwargs):
             :class:`KeyVaultBackupResult` returned by :func:`begin_backup`, for example
             https://<account>.blob.core.windows.net/backup/mhsm-account-2020090117323313
         :param str sas_token: a Shared Access Signature (SAS) token authorizing access to the blob storage resource
+
         :keyword str continuation_token: a continuation token to restart polling from a saved state
         :keyword str key_name: name of a single key in the backup. When set, only this key will be restored.
+
         :rtype: ~azure.core.polling.LROPoller
 
         Examples:

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

@@ -23,21 +23,21 @@
 _VaultId = namedtuple("_VaultId", ["vault_url", "collection", "name", "version"])
 
 
-def parse_vault_id(url):
+def parse_vault_id(url: str) -> "_VaultId":
     try:
         parsed_uri = urlparse(url)
     except Exception:  # pylint: disable=broad-except
-        raise ValueError("'{}' is not a valid url".format(url))
+        raise ValueError(f"'{url}' is not a valid url")
     if not (parsed_uri.scheme and parsed_uri.hostname):
-        raise ValueError("'{}' is not a valid url".format(url))
+        raise ValueError(f"'{url}' is not a valid url")
 
     path = list(filter(None, parsed_uri.path.split("/")))
 
     if len(path) < 2 or len(path) > 3:
-        raise ValueError("'{}' is not a valid vault url".format(url))
+        raise ValueError(f"'{url}' is not a valid vault url")
 
     return _VaultId(
-        vault_url="{}://{}".format(parsed_uri.scheme, parsed_uri.hostname),
+        vault_url=f"{parsed_uri.scheme}://{parsed_uri.hostname}",
         collection=path[0],
         name=path[1],
         version=path[2] if len(path) == 3 else None,
@@ -47,8 +47,7 @@ def parse_vault_id(url):
 BackupLocation = namedtuple("BackupLocation", ["container_url", "folder_name"])
 
 
-def parse_folder_url(folder_url):
-    # type: (str) -> BackupLocation
+def parse_folder_url(folder_url: str) -> "BackupLocation":
     """Parse the blob container URL and folder name from a backup's blob storage URL.
 
     For example, https://<account>.blob.core.windows.net/backup/mhsm-account-2020090117323313 parses to
@@ -66,7 +65,7 @@ def parse_folder_url(folder_url):
         folder_name = stripped_path[len(container) + 1 :]
 
         # this intentionally discards any SAS token in the URL--methods require the SAS token as a separate parameter
-        container_url = "{}://{}/{}".format(parsed.scheme, parsed.netloc, container)
+        container_url = f"{parsed.scheme}://{parsed.netloc}/{container}"
 
         return BackupLocation(container_url, folder_name)
     except:  # pylint:disable=broad-except

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

@@ -33,7 +33,7 @@
 class AsyncChallengeAuthPolicy(AsyncBearerTokenCredentialPolicy):
     """policy for handling HTTP authentication challenges"""
 
-    def __init__(self, credential: "AsyncTokenCredential", *scopes: str, **kwargs: "Any") -> None:
+    def __init__(self, credential: "AsyncTokenCredential", *scopes: str, **kwargs) -> None:
         super().__init__(credential, *scopes, **kwargs)
         self._credential = credential
         self._token = None  # type: Optional[AccessToken]
@@ -50,7 +50,7 @@ async def on_request(self, request: "PipelineRequest") -> None:
                 self._token = await self._credential.get_token(scope, tenant_id=challenge.tenant_id)
 
             # ignore mypy's warning -- although self._token is Optional, get_token raises when it fails to get a token
-            request.http_request.headers["Authorization"] = "Bearer {}".format(self._token.token)  # type: ignore
+            request.http_request.headers["Authorization"] = f"Bearer {self._token.token}"  # type: ignore
             return
 
         # else: discover authentication information by eliciting a challenge from Key Vault. Remove any request data,

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

@@ -20,7 +20,7 @@
 
 
 class AsyncKeyVaultClientBase(object):
-    def __init__(self, vault_url: str, credential: "AsyncTokenCredential", **kwargs: "Any") -> None:
+    def __init__(self, vault_url: str, credential: "AsyncTokenCredential", **kwargs) -> None:
         if not credential:
             raise ValueError(
                 "credential should be an object supporting the AsyncTokenCredential protocol, "
@@ -63,8 +63,8 @@ def __init__(self, vault_url: str, credential: "AsyncTokenCredential", **kwargs:
             self._models = _KeyVaultClient.models(api_version=api_version)
         except ValueError:
             raise NotImplementedError(
-                "This package doesn't support API version '{}'. ".format(api_version)
-                + "Supported versions: {}".format(", ".join(v.value for v in ApiVersion))
+                f"This package doesn't support API version '{api_version}'. "
+                + f"Supported versions: {', '.join(v.value for v in ApiVersion)}"
             )
 
     @property

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

@@ -35,17 +35,15 @@
     from azure.core.pipeline import PipelineResponse
 
 
-def _enforce_tls(request):
-    # type: (PipelineRequest) -> None
+def _enforce_tls(request: PipelineRequest) -> None:
     if not request.http_request.url.lower().startswith("https"):
         raise ServiceRequestError(
             "Bearer token authentication is not permitted for non-TLS protected (non-https) URLs."
         )
 
 
-def _update_challenge(request, challenger):
-    # type: (PipelineRequest, PipelineResponse) -> HttpChallenge
-    """parse challenge from challenger, cache it, return it"""
+def _update_challenge(request: PipelineRequest, challenger: "PipelineResponse") -> HttpChallenge:
+    """Parse challenge from challenger, cache it, return it"""
 
     challenge = HttpChallenge(
         request.http_request.url,
@@ -57,17 +55,15 @@ def _update_challenge(request, challenger):
 
 
 class ChallengeAuthPolicy(BearerTokenCredentialPolicy):
-    """policy for handling HTTP authentication challenges"""
+    """Policy for handling HTTP authentication challenges"""
 
-    def __init__(self, credential, *scopes, **kwargs):
-        # type: (TokenCredential, *str, **Any) -> None
+    def __init__(self, credential: "TokenCredential", *scopes: str, **kwargs) -> None:
         super(ChallengeAuthPolicy, self).__init__(credential, *scopes, **kwargs)
         self._credential = credential
         self._token = None  # type: Optional[AccessToken]
         self._verify_challenge_resource = kwargs.pop("verify_challenge_resource", True)
 
-    def on_request(self, request):
-        # type: (PipelineRequest) -> None
+    def on_request(self, request: PipelineRequest) -> None:
         _enforce_tls(request)
         challenge = ChallengeCache.get_challenge_for_url(request.http_request.url)
         if challenge:
@@ -78,7 +74,7 @@ def on_request(self, request):
                 self._token = self._credential.get_token(scope, tenant_id=challenge.tenant_id)
 
             # ignore mypy's warning -- although self._token is Optional, get_token raises when it fails to get a token
-            request.http_request.headers["Authorization"] = "Bearer {}".format(self._token.token)  # type: ignore
+            request.http_request.headers["Authorization"] = f"Bearer {self._token.token}"  # type: ignore
             return
 
         # else: discover authentication information by eliciting a challenge from Key Vault. Remove any request data,
@@ -90,8 +86,7 @@ def on_request(self, request):
             request.http_request.set_json_body(None)
             request.http_request.headers["Content-Length"] = "0"
 
-    def on_challenge(self, request, response):
-        # type: (PipelineRequest, 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
@@ -119,6 +114,5 @@ def on_challenge(self, request, response):
         return True
 
     @property
-    def _need_new_token(self):
-        # type: () -> bool
+    def _need_new_token(self) -> bool:
         return not self._token or self._token.expires_on - time.time() < 300