minor changes

Author: Pulkit0110

google/cloud/storage/_experimental/asyncio/_utils.py

@@ -33,3 +33,8 @@ def raise_if_no_fast_crc32c():
             "C extension is required for faster data integrity checks."
             "For more information, see https://github.com/googleapis/python-crc32c."
         )
+
+def update_write_handle_if_exists(obj, response):
+    """Update the write_handle attribute of an object if it exists in the response."""
+    if hasattr(response, "write_handle") and response.write_handle is not None:
+        obj.write_handle = response.write_handle

google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py

@@ -32,7 +32,7 @@ class _AsyncAbstractObjectStream(abc.ABC):
     :param generation_number: (Optional) If present, selects a specific revision of
                               this object.
 
-    :type handle: bytes
+    :type handle: Any
     :param handle: (Optional) The handle for the object, could be read_handle or
                    write_handle, based on how the stream is used.
     """
@@ -42,13 +42,13 @@ def __init__(
         bucket_name: str,
         object_name: str,
         generation_number: Optional[int] = None,
-        handle: Optional[bytes] = None,
+        handle: Optional[Any] = None,
     ) -> None:
         super().__init__()
         self.bucket_name: str = bucket_name
         self.object_name: str = object_name
         self.generation_number: Optional[int] = generation_number
-        self.handle: Optional[bytes] = handle
+        self.handle: Optional[Any] = handle
 
     @abc.abstractmethod
     async def open(self) -> None:

google/cloud/storage/_experimental/asyncio/async_appendable_object_writer.py

@@ -36,7 +36,7 @@
 from google.cloud._storage_v2.types.storage import BidiWriteObjectRequest
 
 
-from ._utils import raise_if_no_fast_crc32c
+from . import _utils
 from google.cloud import _storage_v2
 from google.cloud.storage._experimental.asyncio.async_grpc_client import (
     AsyncGrpcClient,
@@ -117,8 +117,8 @@ def __init__(
         client: AsyncGrpcClient.grpc_client,
         bucket_name: str,
         object_name: str,
-        generation=None,
-        write_handle=None,
+        generation: Optional[int] = None,
+        write_handle: Optional[_storage_v2.BidiWriteHandle] = None,
         writer_options: Optional[dict] = None,
     ):
         """
@@ -164,7 +164,7 @@ def __init__(
         :type object_name: str
         :param object_name: The name of the GCS Appendable Object to be written.
 
-        :type generation: int
+        :type generation: Optional[int]
         :param generation: (Optional) If present, creates writer for that
             specific revision of that object. Use this to append data to an
             existing Appendable Object.
@@ -174,10 +174,10 @@ def __init__(
             overwriting existing objects).
 
             Warning: If `None`, a new object is created. If an object with the
-            same name already exists, it will be overwritten the moment 
+            same name already exists, it will be overwritten the moment
             `writer.open()` is called.
 
-        :type write_handle: bytes
+        :type write_handle: _storage_v2.BidiWriteHandle
         :param write_handle: (Optional) An handle for writing the object.
             If provided, opening the bidi-gRPC connection will be faster.
 
@@ -189,7 +189,7 @@ def __init__(
                 servers. Default is `_DEFAULT_FLUSH_INTERVAL_BYTES`.
                 Must be a multiple of `_MAX_CHUNK_SIZE_BYTES`.
         """
-        raise_if_no_fast_crc32c()
+        _utils.raise_if_no_fast_crc32c()
         self.client = client
         self.bucket_name = bucket_name
         self.object_name = object_name

google/cloud/storage/_experimental/asyncio/async_multi_range_downloader.py

@@ -129,7 +129,7 @@ async def create_mrd(
         bucket_name: str,
         object_name: str,
         generation_number: Optional[int] = None,
-        read_handle: Optional[bytes] = None,
+        read_handle: Optional[_storage_v2.BidiReadHandle] = None,
         retry_policy: Optional[AsyncRetry] = None,
         metadata: Optional[List[Tuple[str, str]]] = None,
     ) -> AsyncMultiRangeDownloader:
@@ -149,7 +149,7 @@ async def create_mrd(
         :param generation_number: (Optional) If present, selects a specific
                                   revision of this object.
 
-        :type read_handle: bytes
+        :type read_handle: _storage_v2.BidiReadHandle
         :param read_handle: (Optional) An existing handle for reading the object.
                             If provided, opening the bidi-gRPC connection will be faster.
 
@@ -172,7 +172,7 @@ def __init__(
         bucket_name: str,
         object_name: str,
         generation_number: Optional[int] = None,
-        read_handle: Optional[bytes] = None,
+        read_handle: Optional[_storage_v2.BidiReadHandle] = None,
     ) -> None:
         """Constructor for AsyncMultiRangeDownloader, clients are not adviced to
          use it directly. Instead it's adviced to use the classmethod `create_mrd`.
@@ -190,7 +190,7 @@ def __init__(
         :param generation_number: (Optional) If present, selects a specific revision of
                                   this object.
 
-        :type read_handle: bytes
+        :type read_handle: _storage_v2.BidiReadHandle
         :param read_handle: (Optional) An existing read handle.
         """
 
@@ -200,7 +200,7 @@ def __init__(
         self.bucket_name = bucket_name
         self.object_name = object_name
         self.generation_number = generation_number
-        self.read_handle = read_handle
+        self.read_handle: Optional[_storage_v2.BidiReadHandle] = read_handle
         self.read_obj_str: Optional[_AsyncReadObjectStream] = None
         self._is_stream_open: bool = False
         self._routing_token: Optional[str] = None

google/cloud/storage/_experimental/asyncio/async_read_object_stream.py

@@ -51,7 +51,7 @@ class _AsyncReadObjectStream(_AsyncAbstractObjectStream):
     :param generation_number: (Optional) If present, selects a specific revision of
                               this object.
 
-    :type read_handle: bytes
+    :type read_handle: _storage_v2.BidiReadHandle
     :param read_handle: (Optional) An existing handle for reading the object.
                         If provided, opening the bidi-gRPC connection will be faster.
     """
@@ -62,7 +62,7 @@ def __init__(
         bucket_name: str,
         object_name: str,
         generation_number: Optional[int] = None,
-        read_handle: Optional[bytes] = None,
+        read_handle: Optional[_storage_v2.BidiReadHandle] = None,
     ) -> None:
         if client is None:
             raise ValueError("client must be provided")
@@ -77,7 +77,7 @@ def __init__(
             generation_number=generation_number,
         )
         self.client: AsyncGrpcClient.grpc_client = client
-        self.read_handle: Optional[bytes] = read_handle
+        self.read_handle: Optional[_storage_v2.BidiReadHandle] = read_handle
 
         self._full_bucket_name = f"projects/_/buckets/{self.bucket_name}"
 

tests/perf/microbenchmarks/README.md

@@ -15,10 +15,13 @@ pytest --benchmark-json=output.json -vv -s tests/perf/microbenchmarks/reads/test
 
 To run a single test, append `::` followed by the test name to the file path.
 
-Example:
+Examples:
 ```bash
 pytest --benchmark-json=output.json -vv -s tests/perf/microbenchmarks/reads/test_reads.py::test_downloads_single_proc_single_coro
 ```
+```bash
+pytest --benchmark-json=output.json -vv -s tests/perf/microbenchmarks/writes/test_writes.py::test_uploads_single_proc_single_coro
+```
 
 ## Configuration
 

tests/perf/microbenchmarks/conftest.py

@@ -26,6 +26,7 @@
     AsyncAppendableObjectWriter,
 )
 from google.cloud.storage._experimental.asyncio.async_grpc_client import AsyncGrpcClient
+from tests.perf.microbenchmarks.writes.parameters import WriteParameters
 
 _OBJECT_NAME_PREFIX = "micro-benchmark"
 
@@ -72,7 +73,7 @@ def publish_resource_metrics(benchmark: Any, monitor: ResourceMonitor) -> None:
 
 async def upload_appendable_object(bucket_name, object_name, object_size, chunk_size):
     # flush interval set to little over 1GiB to minimize number of flushes.
-    # this method is to write "appendable" objects which will be used for 
+    # this method is to write "appendable" objects which will be used for
     # benchmarking reads, hence not concerned performance of writes here.
     writer = AsyncAppendableObjectWriter(
         AsyncGrpcClient().grpc_client, bucket_name, object_name, writer_options={"FLUSH_INTERVAL_BYTES": 1026 * 1024 ** 2}
@@ -135,10 +136,16 @@ def _create_files(num_files, bucket_name, bucket_type, object_size, chunk_size=1
 @pytest.fixture
 def workload_params(request):
     params = request.param
-    files_names = _create_files(
-        params.num_files,
-        params.bucket_name,
-        params.bucket_type,
-        params.file_size_bytes,
-    )
-    return params, files_names
+    if isinstance(params, WriteParameters):
+        files_names = [
+            f"{_OBJECT_NAME_PREFIX}-{uuid.uuid4().hex[:5]}"
+            for _ in range(params.num_files)
+        ]
+    else:
+        files_names = _create_files(
+            params.num_files,
+            params.bucket_name,
+            params.bucket_type,
+            params.file_size_bytes,
+        )
+    return params, files_names