[Storage] [Named Keywords] [Blob] _blob_client.py and aio

sdk/storage/azure-storage-blob/azure/storage/blob/_blob_service_client.py

@@ -121,14 +121,17 @@ class BlobServiceClient(StorageAccountHostsMixin, StorageEncryptionMixin):
     def __init__(
         self, account_url: str,
         credential: Optional[Union[str, Dict[str, str], "AzureNamedKeyCredential", "AzureSasCredential", "TokenCredential"]] = None,  # pylint: disable=line-too-long
+        *,
+        api_version: Optional[str] = None,
+        # TODO
         **kwargs: Any
     ) -> None:
         parsed_url, sas_token = _parse_url(account_url=account_url)
         _, sas_token = parse_query(parsed_url.query)
         self._query_str, credential = self._format_query_string(sas_token, credential)
         super(BlobServiceClient, self).__init__(parsed_url, service='blob', credential=credential, **kwargs)
         self._client = AzureBlobStorage(self.url, base_url=self.url, pipeline=self._pipeline)
-        self._client._config.version = get_api_version(kwargs)  # type: ignore [assignment]
+        self._client._config.version = get_api_version(api_version)  # type: ignore [assignment]
         self._configure_encryption(kwargs)
 
     def _format_url(self, hostname):
@@ -240,7 +243,7 @@ def get_account_information(self, **kwargs: Any) -> Dict[str, str]:
                 :caption: Getting account information for the blob service.
         """
         try:
-            return self._client.service.get_account_info(cls=return_response_headers, **kwargs) # type: ignore
+            return self._client.service.get_account_info(cls=return_response_headers, **kwargs)  # type: ignore
         except HttpResponseError as error:
             process_storage_error(error)
 
@@ -284,7 +287,7 @@ def get_service_stats(self, **kwargs: Any) -> Dict[str, Any]:
         """
         timeout = kwargs.pop('timeout', None)
         try:
-            stats = self._client.service.get_statistics( # type: ignore
+            stats = self._client.service.get_statistics(  # type: ignore
                 timeout=timeout, use_location=LocationMode.SECONDARY, **kwargs)
             return service_stats_deserialize(stats)
         except HttpResponseError as error:
@@ -391,7 +394,7 @@ def set_service_properties(
             logging=analytics_logging,
             hour_metrics=hour_metrics,
             minute_metrics=minute_metrics,
-            cors=CorsRule._to_generated(cors), # pylint: disable=protected-access
+            cors=CorsRule._to_generated(cors),  # pylint: disable=protected-access
             default_service_version=target_version,
             delete_retention_policy=delete_retention_policy,
             static_website=static_website
@@ -648,7 +651,7 @@ def _rename_container(self, name: str, new_name: str, **kwargs: Any) -> Containe
         except AttributeError:
             kwargs['source_lease_id'] = lease
         try:
-            renamed_container._client.container.rename(name, **kwargs)  # pylint: disable = protected-access
+            renamed_container._client.container.rename(name, **kwargs)  # pylint: disable=protected-access
             return renamed_container
         except HttpResponseError as error:
             process_storage_error(error)
@@ -685,7 +688,7 @@ def undelete_container(
             warnings.warn("`new_name` is no longer supported.", DeprecationWarning)
         container = self.get_container_client(new_name or deleted_container_name)
         try:
-            container._client.container.restore(deleted_container_name=deleted_container_name,  # pylint: disable = protected-access
+            container._client.container.restore(deleted_container_name=deleted_container_name,  # pylint: disable=protected-access
                                                 deleted_container_version=deleted_container_version,
                                                 timeout=kwargs.pop('timeout', None), **kwargs)
             return container
@@ -718,8 +721,8 @@ def get_container_client(self, container: Union[ContainerProperties, str]) -> Co
         else:
             container_name = container
         _pipeline = Pipeline(
-            transport=TransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
-            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+            transport=TransportWrapper(self._pipeline._transport),  # pylint: disable=protected-access
+            policies=self._pipeline._impl_policies  # pylint: disable=protected-access
         )
         return ContainerClient(
             self.url, container_name=container_name,
@@ -776,8 +779,8 @@ def get_blob_client(
         else:
             container_name = container
         _pipeline = Pipeline(
-            transport=TransportWrapper(self._pipeline._transport), # pylint: disable = protected-access
-            policies=self._pipeline._impl_policies # pylint: disable = protected-access
+            transport=TransportWrapper(self._pipeline._transport),  # pylint: disable=protected-access
+            policies=self._pipeline._impl_policies  # pylint: disable=protected-access
         )
         return BlobClient(
             self.url, container_name=container_name, blob_name=blob_name, snapshot=snapshot,

sdk/storage/azure-storage-blob/azure/storage/blob/_container_client.py

@@ -137,6 +137,9 @@ def __init__(
         self, account_url: str,
         container_name: str,
         credential: Optional[Union[str, Dict[str, str], "AzureNamedKeyCredential", "AzureSasCredential", "TokenCredential"]] = None,  # pylint: disable=line-too-long
+        *,
+        api_version: Optional[str] = None,
+        # TODO
         **kwargs: Any
     ) -> None:
         parsed_url, sas_token = _parse_url(account_url=account_url, container_name=container_name)
@@ -146,7 +149,7 @@ def __init__(
         self._raw_credential = credential if credential else sas_token
         self._query_str, credential = self._format_query_string(sas_token, credential)
         super(ContainerClient, self).__init__(parsed_url, service='blob', credential=credential, **kwargs)
-        self._api_version = get_api_version(kwargs)
+        self._api_version = get_api_version(api_version)
         self._client = self._build_generated_client()
         self._configure_encryption(kwargs)
 

sdk/storage/azure-storage-blob/azure/storage/blob/_serialize.py

@@ -31,6 +31,7 @@
 
 if TYPE_CHECKING:
     from ._lease import BlobLeaseClient
+    from aio._lease_async import BlobLeaseClient as AsyncBlobLeaseClient
 
 
 _SUPPORTED_API_VERSIONS = [
@@ -90,12 +91,16 @@ def _get_match_headers(
     return if_match, if_none_match
 
 
-def get_access_conditions(lease: Optional[Union["BlobLeaseClient", str]]) -> Optional[LeaseAccessConditions]:
-    try:
-        lease_id = lease.id # type: ignore
-    except AttributeError:
-        lease_id = lease # type: ignore
-    return LeaseAccessConditions(lease_id=lease_id) if lease_id else None
+def get_access_conditions(
+    lease: Optional[Union["BlobLeaseClient", "AsyncBlobLeaseClient", str]]
+) -> Optional[LeaseAccessConditions]:
+    if lease is None:
+        return None
+    if hasattr(lease, "id"):
+        lease_id = lease.id  # type: ignore
+    else:
+        lease_id = lease  # type: ignore
+    return LeaseAccessConditions(lease_id=lease_id)
 
 
 def get_modify_conditions(kwargs: Dict[str, Any]) -> ModifiedAccessConditions:
@@ -143,18 +148,19 @@ def get_container_cpk_scope_info(kwargs: Dict[str, Any]) -> Optional[ContainerCp
     return None
 
 
-def get_api_version(kwargs: Dict[str, Any]) -> str:
-    api_version = kwargs.get('api_version', None)
+def get_api_version(api_version: Optional[str] = None) -> str:
     if api_version and api_version not in _SUPPORTED_API_VERSIONS:
         versions = '\n'.join(_SUPPORTED_API_VERSIONS)
         raise ValueError(f"Unsupported API version '{api_version}'. Please select from:\n{versions}")
     return api_version or _SUPPORTED_API_VERSIONS[-1]
 
+
 def get_version_id(self_vid: Optional[str], kwargs: Dict[str, Any]) -> Optional[str]:
     if 'version_id' in kwargs:
         return cast(str, kwargs.pop('version_id'))
     return self_vid
 
+
 def serialize_blob_tags_header(tags: Optional[Dict[str, str]] = None) -> Optional[str]:
     if tags is None:
         return None