[Storage] Enabled black for azure-storage-blob-changefeed package (#47049)

Author: weirongw23-msft

sdk/storage/azure-storage-blob-changefeed/azure/__init__.py

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

sdk/storage/azure-storage-blob-changefeed/azure/storage/__init__.py

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

sdk/storage/azure-storage-blob-changefeed/azure/storage/blob/__init__.py

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

sdk/storage/azure-storage-blob-changefeed/azure/storage/blob/changefeed/__init__.py

@@ -9,6 +9,4 @@
 
 __version__ = VERSION
 
-__all__ = [
-    'ChangeFeedClient'
-]
+__all__ = ["ChangeFeedClient"]

sdk/storage/azure-storage-blob-changefeed/azure/storage/blob/changefeed/_change_feed_client.py

@@ -5,10 +5,7 @@
 # --------------------------------------------------------------------------
 # pylint: disable=docstring-keyword-should-match-keyword-only
 
-from typing import (
-    Any, Dict, Optional, Union,
-    TYPE_CHECKING
-)
+from typing import Any, Dict, Optional, Union, TYPE_CHECKING
 from typing_extensions import Self
 
 from azure.core.paging import ItemPaged
@@ -60,17 +57,24 @@ class ChangeFeedClient:
             :dedent: 8
             :caption: Creating the ChangeFeedClient from a URL to a public blob (no auth needed).
     """
+
     def __init__(
-        self, account_url: str,
-        credential: Optional[Union[str, Dict[str, str], "AzureNamedKeyCredential", "AzureSasCredential", "TokenCredential"]] = None,  # pylint: disable=line-too-long
+        self,
+        account_url: str,
+        credential: Optional[
+            Union[str, Dict[str, str], "AzureNamedKeyCredential", "AzureSasCredential", "TokenCredential"]
+        ] = None,
         **kwargs: Any
     ) -> None:
         self._blob_service_client = BlobServiceClient(account_url, credential, **kwargs)
 
     @classmethod
     def from_connection_string(
-        cls, conn_str: str,
-        credential: Optional[Union[str, Dict[str, str], "AzureNamedKeyCredential", "AzureSasCredential", "TokenCredential"]] = None,  # pylint: disable=line-too-long
+        cls,
+        conn_str: str,
+        credential: Optional[
+            Union[str, Dict[str, str], "AzureNamedKeyCredential", "AzureSasCredential", "TokenCredential"]
+        ] = None,
         **kwargs: Any
     ) -> Self:
         """Create ChangeFeedClient from a Connection String.
@@ -94,9 +98,9 @@ def from_connection_string(
         :returns: A change feed client.
         :rtype: ~azure.storage.blob.changefeed.ChangeFeedClient
         """
-        account_url, secondary, credential = parse_connection_str(conn_str, credential, 'blob')
-        if 'secondary_hostname' not in kwargs:
-            kwargs['secondary_hostname'] = secondary
+        account_url, secondary, credential = parse_connection_str(conn_str, credential, "blob")
+        if "secondary_hostname" not in kwargs:
+            kwargs["secondary_hostname"] = secondary
         return cls(account_url, credential=credential, **kwargs)
 
     @distributed_trace
@@ -130,11 +134,8 @@ def list_changes(self, **kwargs: Any) -> ItemPaged[Dict[str, Any]]:
                 :dedent: 8
                 :caption: List change feed events by page.
         """
-        results_per_page = kwargs.pop('results_per_page', None)
+        results_per_page = kwargs.pop("results_per_page", None)
         container_client = self._blob_service_client.get_container_client("$blobchangefeed")
         return ItemPaged(
-            container_client,
-            results_per_page=results_per_page,
-            page_iterator_class=ChangeFeedPaged,
-            **kwargs
+            container_client, results_per_page=results_per_page, page_iterator_class=ChangeFeedPaged, **kwargs
         )

sdk/storage/azure-storage-blob-changefeed/azure/storage/blob/changefeed/_models.py

@@ -9,10 +9,7 @@
 import copy
 import json
 from datetime import datetime
-from typing import (
-    Any, Dict, Iterator, List, Optional, Union,
-    TYPE_CHECKING
-)
+from typing import Any, Dict, Iterator, List, Optional, Union, TYPE_CHECKING
 from typing_extensions import Self
 
 from azure.core.exceptions import HttpResponseError
@@ -35,10 +32,11 @@
 
 class Segment:
     def __init__(
-        self, client: Union["ContainerClient", "AsyncContainerClient"],
+        self,
+        client: Union["ContainerClient", "AsyncContainerClient"],
         segment_path: str,
         page_size: int,
-        segment_cursor: Optional[Dict[str, Any]] = None
+        segment_cursor: Optional[Dict[str, Any]] = None,
     ) -> None:
         self.client = client
         self.segment_path = segment_path
@@ -91,8 +89,9 @@ def _initialize(self, segment_cursor=None):
                 self.shards.append(Shard(self.client, shard_path))
         else:
             start_shard_path = segment_cursor["CurrentShardPath"]
-            shard_cursors = {shard_cursor["CurrentChunkPath"][:-10]: shard_cursor
-                             for shard_cursor in segment_cursor["ShardCursors"]}
+            shard_cursors = {
+                shard_cursor["CurrentChunkPath"][:-10]: shard_cursor for shard_cursor in segment_cursor["ShardCursors"]
+            }
 
             if shard_paths:
                 # Initialize all shards using the shard cursors
@@ -129,30 +128,33 @@ class ChangeFeedPaged(PageIterator):
     """The current page of listed results."""
 
     def __init__(
-        self, container_client: Union["ContainerClient", "AsyncContainerClient"],
+        self,
+        container_client: Union["ContainerClient", "AsyncContainerClient"],
         results_per_page: Optional[int] = None,
         start_time: Optional[datetime] = None,
         end_time: Optional[datetime] = None,
-        continuation_token: Optional[str] = None
+        continuation_token: Optional[str] = None,
     ) -> None:
         if (start_time or end_time) and continuation_token:
             raise ValueError("start_time/end_time and continuation_token shouldn't be specified at the same time")
         super(ChangeFeedPaged, self).__init__(
-            get_next=self._get_next_cf,
-            extract_data=self._extract_data_cb,
-            continuation_token=continuation_token or ""
+            get_next=self._get_next_cf, extract_data=self._extract_data_cb, continuation_token=continuation_token or ""
         )
         dict_continuation_token = json.loads(continuation_token) if continuation_token else None
 
-        if dict_continuation_token and (container_client.primary_hostname != dict_continuation_token["UrlHost"]):  # pylint: disable=unsubscriptable-object
+        if dict_continuation_token and (container_client.primary_hostname != dict_continuation_token["UrlHost"]):
             raise ValueError("The token is not for the current storage account.")
-        if dict_continuation_token and (dict_continuation_token["CursorVersion"] != 1):  # pylint: disable=unsubscriptable-object
+        if dict_continuation_token and (dict_continuation_token["CursorVersion"] != 1):
             raise ValueError("The CursorVersion is not supported by the current SDK.")
         self.results_per_page = results_per_page or 5000
         self.current_page = None
-        self._change_feed = ChangeFeed(container_client, self.results_per_page, start_time=start_time,
-                                       end_time=end_time,
-                                       cf_cursor=dict_continuation_token)
+        self._change_feed = ChangeFeed(
+            container_client,
+            self.results_per_page,
+            start_time=start_time,
+            end_time=end_time,
+            cf_cursor=dict_continuation_token,
+        )
 
     def _get_next_cf(self, continuation_token):  # pylint:disable=inconsistent-return-statements, unused-argument
         try:
@@ -174,11 +176,12 @@ def _extract_data_cb(self, event_list):
 
 class ChangeFeed:
     def __init__(
-        self, client: Union["ContainerClient", "AsyncContainerClient"],
+        self,
+        client: Union["ContainerClient", "AsyncContainerClient"],
         page_size: int,
         start_time: Optional[datetime] = None,
         end_time: Optional[datetime] = None,
-        cf_cursor: Optional[Dict[str, Any]] = None
+        cf_cursor: Optional[Dict[str, Any]] = None,
     ) -> None:
         self.client = client
         self.page_size = page_size
@@ -189,16 +192,19 @@ def __init__(
         # the end time is in str format
         end_time_in_cursor = cf_cursor["EndTime"] if cf_cursor else None
         # convert the end time in str format to a datetime object
-        end_time_in_cursor_obj = \
+        end_time_in_cursor_obj = (
             datetime.strptime(end_time_in_cursor, "%Y-%m-%dT%H:%M:%S+00:00") if end_time_in_cursor else None
+        )
         # self.end_time is in datetime format
         self.end_time = end_time or end_time_in_cursor_obj
 
         cur_segment_cursor = cf_cursor["CurrentSegmentCursor"] if cf_cursor else None
 
-        self.cursor = {"CursorVersion": 1,
-                       "EndTime": self.end_time.strftime("%Y-%m-%dT%H:%M:%S+00:00") if self.end_time else "",
-                       "UrlHost": self.client.primary_hostname}
+        self.cursor = {
+            "CursorVersion": 1,
+            "EndTime": self.end_time.strftime("%Y-%m-%dT%H:%M:%S+00:00") if self.end_time else "",
+            "UrlHost": self.client.primary_hostname,
+        }
         self._initialize(cur_segment_cursor=cur_segment_cursor)
 
     def __iter__(self) -> Self:
@@ -257,9 +263,8 @@ def _initialize(self, cur_segment_cursor=None):
                 next_segment_path = next(self._segment_paths_generator)
 
         self.current_segment = self._get_next_segment(
-            next_segment_path,
-            self.page_size,
-            segment_cursor=cur_segment_cursor if cur_segment_cursor else None)
+            next_segment_path, self.page_size, segment_cursor=cur_segment_cursor if cur_segment_cursor else None
+        )
 
     def _get_next_segment(self, segment_path, page_size, segment_cursor=None):
         if segment_path:
@@ -297,23 +302,24 @@ def _parse_datetime_from_segment_path(segment_path):
 
     def _is_earlier_than_start_time(self, segment_path):
         segment_date = self._parse_datetime_from_segment_path(segment_path)
-        opaque_start_date = datetime(self.start_time.year, self.start_time.month,
-                                     self.start_time.day, self.start_time.hour)
+        opaque_start_date = datetime(
+            self.start_time.year, self.start_time.month, self.start_time.day, self.start_time.hour
+        )
 
         return segment_date < opaque_start_date
 
     def _is_later_than_end_time(self, segment_path):
         segment_date = self._parse_datetime_from_segment_path(segment_path)
-        opaque_end_date = datetime(self.end_time.year, self.end_time.month,
-                                   self.end_time.day, self.end_time.hour)
+        opaque_end_date = datetime(self.end_time.year, self.end_time.month, self.end_time.day, self.end_time.hour)
         return segment_date > opaque_end_date
 
 
 class Shard:
     def __init__(
-        self, client: Union["ContainerClient", "AsyncContainerClient"],
+        self,
+        client: Union["ContainerClient", "AsyncContainerClient"],
         shard_path: str,
-        shard_cursor: Optional[Dict[str, Any]] = None
+        shard_cursor: Optional[Dict[str, Any]] = None,
     ) -> None:
         self.client = client
         self.shard_path = shard_path
@@ -348,8 +354,9 @@ def _initialize(self, shard_cursor=None):
 
         # move cursor to the expected chunk
         if shard_cursor:
-            while self.unprocessed_chunk_path_props and \
-                    self.unprocessed_chunk_path_props[0].name != shard_cursor.get("CurrentChunkPath"):
+            while self.unprocessed_chunk_path_props and self.unprocessed_chunk_path_props[0].name != shard_cursor.get(
+                "CurrentChunkPath"
+            ):
                 self.unprocessed_chunk_path_props.popleft()
             self.current_chunk = self._get_next_chunk(chunk_cursor=shard_cursor)
         else:
@@ -365,10 +372,7 @@ def _get_next_chunk(self, chunk_cursor=None):
 class ChangeFeedStreamer:
     """File-like streaming iterator."""
 
-    def __init__(
-        self, blob_client: "BlobClient",
-        chunk_file_start: int = 0
-    ) -> None:
+    def __init__(self, blob_client: "BlobClient", chunk_file_start: int = 0) -> None:
         self._chunk_file_start = chunk_file_start or 0  # this value will never be updated
         self._download_offset = self._chunk_file_start  # range start of the next download
         self.object_position = self._chunk_file_start  # track the most recently read sync marker position
@@ -378,8 +382,9 @@ def __init__(
         self._buf_start = self._chunk_file_start  # the start position of the chunk file to buffer
         self._chunk_size_snapshot = blob_client.get_blob_properties().size
         length = self._chunk_size_snapshot - self._chunk_file_start
-        self._iterator = blob_client.download_blob(offset=self._chunk_file_start,
-                                                   length=length).chunks() if length > 0 else iter([])
+        self._iterator = (
+            blob_client.download_blob(offset=self._chunk_file_start, length=length).chunks() if length > 0 else iter([])
+        )
 
     def __len__(self) -> int:
         return self._download_offset
@@ -430,7 +435,7 @@ def read(self, size: int) -> bytes:
         if relative_start < 0:
             raise ValueError("Buffer has dumped too much data")
         relative_end = relative_start + size
-        data = self._buf[relative_start: relative_end]
+        data = self._buf[relative_start:relative_end]
 
         # dump the extra data in buffer
         # buffer start--------------------16bytes----current read position
@@ -449,9 +454,10 @@ def set_object_index(self, event_index: int) -> None:
 
 class Chunk:
     def __init__(
-        self, client: Union["ContainerClient", "AsyncContainerClient"],
+        self,
+        client: Union["ContainerClient", "AsyncContainerClient"],
         chunk_path: str,
-        chunk_cursor: Optional[Dict[str, Any]] = None
+        chunk_cursor: Optional[Dict[str, Any]] = None,
     ) -> None:
         self.client = client
         self.chunk_path = chunk_path

sdk/storage/azure-storage-blob-changefeed/pyproject.toml

@@ -2,4 +2,4 @@
 mypy = true
 pyright = false
 type_check_samples = true
-black = false
+black = true

sdk/storage/azure-storage-blob-changefeed/samples/change_feed_samples.py

@@ -37,8 +37,9 @@ def list_events_by_page(self):
         # Instantiate a ChangeFeedClient
         # [START list_events_by_page]
         # [START create_change_feed_client]
-        cf_client = ChangeFeedClient("https://{}.blob.core.windows.net".format(self.ACCOUNT_NAME),
-                                     credential=self.ACCOUNT_KEY)
+        cf_client = ChangeFeedClient(
+            "https://{}.blob.core.windows.net".format(self.ACCOUNT_NAME), credential=self.ACCOUNT_KEY
+        )
         # [END create_change_feed_client]
 
         change_feed = cf_client.list_changes(results_per_page=10).by_page()
@@ -56,18 +57,21 @@ def list_events_by_page(self):
 
     def list_all_events(self):
         # [START list_all_events]
-        cf_client = ChangeFeedClient("https://{}.blob.core.windows.net".format(self.ACCOUNT_NAME),
-                                     credential=self.ACCOUNT_KEY)
+        cf_client = ChangeFeedClient(
+            "https://{}.blob.core.windows.net".format(self.ACCOUNT_NAME), credential=self.ACCOUNT_KEY
+        )
         change_feed = cf_client.list_changes()
 
         # print all events
         for event in change_feed:
             print(event)
+
     # [END list_all_events]
 
     def list_range_of_events(self):
-        cf_client = ChangeFeedClient("https://{}.blob.core.windows.net".format(self.ACCOUNT_NAME),
-                                     credential=self.ACCOUNT_KEY)
+        cf_client = ChangeFeedClient(
+            "https://{}.blob.core.windows.net".format(self.ACCOUNT_NAME), credential=self.ACCOUNT_KEY
+        )
         start_time = datetime(2020, 8, 18, 10)
         end_time = datetime(2020, 3, 4)
         change_feed = cf_client.list_changes(start_time=start_time, end_time=end_time)
@@ -78,8 +82,9 @@ def list_range_of_events(self):
 
     def list_events_in_live_mode(self):
         # Instantiate a ChangeFeedClient
-        cf_client = ChangeFeedClient("https://{}.blob.core.windows.net".format(self.ACCOUNT_NAME),
-                                     credential=self.ACCOUNT_KEY)
+        cf_client = ChangeFeedClient(
+            "https://{}.blob.core.windows.net".format(self.ACCOUNT_NAME), credential=self.ACCOUNT_KEY
+        )
         token = None
         while True:
             change_feed = cf_client.list_changes(results_per_page=500).by_page(continuation_token=token)
@@ -93,7 +98,7 @@ def list_events_in_live_mode(self):
             print("continue printing events")
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     sample = ChangeFeedSamples()
     sample.list_events_by_page()
     sample.list_all_events()