remove extra logs

Pulkit0110 · Pulkit0110 · commit 637019e935b7 · 2026-01-21T13:46:25.000Z
diff --git a/google/cloud/storage/_experimental/asyncio/async_appendable_object_writer.py b/google/cloud/storage/_experimental/asyncio/async_appendable_object_writer.py
@@ -265,7 +265,6 @@ async def open(
         :raises ValueError: If the stream is already open.
 
         """
-        print("DEBUG: AsyncAppendableObjectWriter.open() called")
         if self._is_stream_open:
             raise ValueError("Underlying bidi-gRPC stream is already open")
 
@@ -291,7 +290,6 @@ def combined_on_error(exc):
             )
 
         async def _do_open():
-            print("DEBUG: AsyncAppendableObjectWriter.open._do_open() called")
             current_metadata = list(metadata) if metadata else []
 
             # Cleanup stream from previous failed attempt, if any.
@@ -312,7 +310,6 @@ async def _do_open():
                 write_handle=self.write_handle,
                 routing_token=self._routing_token,
             )
-            print(f"DEBUG: AsyncAppendableObjectWriter.open._do_open() - created _AsyncWriteObjectStream with routing_token: {self._routing_token}")
 
             if self._routing_token:
                 current_metadata.append(
@@ -322,7 +319,6 @@ async def _do_open():
             await self.write_obj_stream.open(
                 metadata=current_metadata if metadata else None
             )
-            print("DEBUG: AsyncAppendableObjectWriter.open._do_open() - self.write_obj_stream.open() finished")
 
             if self.write_obj_stream.generation_number:
                 self.generation = self.write_obj_stream.generation_number
@@ -333,10 +329,8 @@ async def _do_open():
 
             self._is_stream_open = True
             self._routing_token = None
-            print(f"DEBUG: AsyncAppendableObjectWriter.open._do_open() - stream opened. persisted_size: {self.persisted_size}, write_handle: {self.write_handle}")
 
         await retry_policy(_do_open)()
-        print("DEBUG: AsyncAppendableObjectWriter.open() finished")
 
     async def append(
         self,
@@ -365,7 +359,6 @@ async def append(
 
         :raises ValueError: If the stream is not open.
         """
-        print(f"AsyncAppendableObjectWriter: append() called with {len(data)} bytes")
         if not self._is_stream_open:
             raise ValueError("Stream is not open. Call open() before append().")
         if not data:
@@ -389,7 +382,6 @@ async def generator():
                 resp = None
                 async with self._lock:
                     write_state = state["write_state"]
-                    print(f"AsyncAppendableObjectWriter: append - attempt {attempt_count}, write_state: {write_state}")
                     # If this is a retry or redirect, we must re-open the stream
                     if attempt_count > 1 or write_state.routing_token:
                         logger.info(
@@ -429,10 +421,8 @@ async def generator():
                         # Check if this is an open/state-lookup request (no checksummed_data)
                         if chunk_req.state_lookup and not chunk_req.checksummed_data:
                             # This is an open request - send it and get response
-                            print(f"AsyncAppendableObjectWriter: append - sending state_lookup request: {chunk_req}")
                             await self.write_obj_stream.send(chunk_req)
                             resp = await self.write_obj_stream.recv()
-                            print(f"AsyncAppendableObjectWriter: append - received state_lookup response")
 
                             # Update state from open response
                             if resp:
@@ -446,12 +436,10 @@ async def generator():
                             continue
 
                         # This is a data request - send it
-                        print(f"AsyncAppendableObjectWriter: append - sending data request:")
                         await self.write_obj_stream.send(chunk_req)
 
                     # Get final response from the last request (which has state_lookup=True)
                     resp = await self.write_obj_stream.recv()
-                    print(f"AsyncAppendableObjectWriter: append - received final response")
 
                     if resp:
                         if resp.persisted_size is not None:
@@ -473,7 +461,6 @@ async def generator():
         write_state.persisted_size = self.persisted_size
         write_state.bytes_sent = self.persisted_size
         write_state.bytes_since_last_flush = self.bytes_appended_since_last_flush
-        print(f"AsyncAppendableObjectWriter: append - initial write_state: {write_state}")
 
         retry_manager = _BidiStreamRetryManager(
             _WriteResumptionStrategy(),
@@ -487,7 +474,6 @@ async def generator():
         self.bytes_appended_since_last_flush = write_state.bytes_since_last_flush
         self.persisted_size = write_state.persisted_size
         self.offset = write_state.persisted_size
-        print(f"AsyncAppendableObjectWriter: append - finished. persisted_size: {self.persisted_size}")
 
     async def simple_flush(self) -> None:
         """Flushes the data to the server.
@@ -550,7 +536,6 @@ async def close(self, finalize_on_close=False) -> Union[int, _storage_v2.Object]
             been called).
 
         """
-        print(f"AsyncAppendableObjectWriter: close() called with finalize_on_close={finalize_on_close}")
         if not self._is_stream_open:
             raise ValueError("Stream is not open. Call open() before close().")
 
@@ -580,19 +565,16 @@ async def finalize(self) -> _storage_v2.Object:
         :raises ValueError: If the stream is not open (i.e., `open()` has not
             been called).
         """
-        print("AsyncAppendableObjectWriter: finalize() called")
         if not self._is_stream_open:
             raise ValueError("Stream is not open. Call open() before finalize().")
 
         await self.write_obj_stream.send(
             _storage_v2.BidiWriteObjectRequest(finish_write=True)
         )
         response = await self.write_obj_stream.recv()
-        print(f"AsyncAppendableObjectWriter: finalize - received response: {response}")
         self.object_resource = response.resource
         self.persisted_size = self.object_resource.size
         await self.write_obj_stream.close()
-        print("AsyncAppendableObjectWriter: finalize - stream closed")
 
         self._is_stream_open = False
         self.offset = None
diff --git a/google/cloud/storage/_experimental/asyncio/async_write_object_stream.py b/google/cloud/storage/_experimental/asyncio/async_write_object_stream.py
@@ -73,7 +73,6 @@ def __init__(
         write_handle: Optional[bytes] = None,
         routing_token: Optional[str] = None,
     ) -> None:
-        print(f"in _AsyncWriteObjectStream __init__: object_name={object_name}")
         if client is None:
             raise ValueError("client must be provided")
         if bucket_name is None:
@@ -115,7 +114,6 @@ async def open(self, metadata: Optional[List[Tuple[str, str]]] = None) -> None:
         :raises google.api_core.exceptions.FailedPrecondition:
             if `generation_number` is 0 and object already exists.
         """
-        print(f"in _AsyncWriteObjectStream open: object_name={self.object_name}")
         if self._is_stream_open:
             raise ValueError("Stream is already open")
 
@@ -146,7 +144,6 @@ async def open(self, metadata: Optional[List[Tuple[str, str]]] = None) -> None:
                     routing_token=self.routing_token if self.routing_token else None,
                 ),
             )
-        print(f"in _AsyncWriteObjectStream open: first_bidi_write_req={self.first_bidi_write_req}")
 
         request_params = [f"bucket={self._full_bucket_name}"]
         other_metadata = []
@@ -167,9 +164,7 @@ async def open(self, metadata: Optional[List[Tuple[str, str]]] = None) -> None:
         )
 
         await self.socket_like_rpc.open()  # this is actually 1 send
-        print(f"in _AsyncWriteObjectStream open: rpc opened")
         response = await self.socket_like_rpc.recv()
-        print(f"in _AsyncWriteObjectStream open: received response:")
         self._is_stream_open = True
 
         if response.persisted_size:
@@ -189,7 +184,6 @@ async def open(self, metadata: Optional[List[Tuple[str, str]]] = None) -> None:
 
     async def close(self) -> None:
         """Closes the bidi-gRPC connection."""
-        print(f"in _AsyncWriteObjectStream close: object_name={self.object_name}")
         if not self._is_stream_open:
             raise ValueError("Stream is not open")
         await self.requests_done()
@@ -198,7 +192,6 @@ async def close(self) -> None:
 
     async def requests_done(self):
         """Signals that all requests have been sent."""
-        print(f"in _AsyncWriteObjectStream requests_done: object_name={self.object_name}")
         await self.socket_like_rpc.send(None)
         await self.socket_like_rpc.recv()
 
@@ -212,7 +205,6 @@ async def send(
                 The request message to send. This is typically used to specify
                 the read offset and limit.
         """
-        print(f"in _AsyncWriteObjectStream send: sending request")
         if not self._is_stream_open:
             raise ValueError("Stream is not open")
         await self.socket_like_rpc.send(bidi_write_object_request)
@@ -229,9 +221,7 @@ async def recv(self) -> _storage_v2.BidiWriteObjectResponse:
         """
         if not self._is_stream_open:
             raise ValueError("Stream is not open")
-        print(f"in _AsyncWriteObjectStream recv: receiving response")
         response = await self.socket_like_rpc.recv()
-        print(f"in _AsyncWriteObjectStream recv: received response")
         # Update write_handle if present in response
         if response:
             if response.write_handle:
diff --git a/google/cloud/storage/_experimental/asyncio/retry/bidi_stream_retry_manager.py b/google/cloud/storage/_experimental/asyncio/retry/bidi_stream_retry_manager.py
@@ -36,7 +36,6 @@ def __init__(
                 bidi operation (e.g., reads or writes).
             send_and_recv: An async callable that opens a new gRPC stream.
         """
-        print("Initializing _BidiStreamRetryManager.")
         self._strategy = strategy
         self._send_and_recv = send_and_recv
 
@@ -48,11 +47,9 @@ async def execute(self, initial_state: Any, retry_policy):
             retry_policy: The `google.api_core.retry.AsyncRetry` object to
                 govern the retry behavior for this specific operation.
         """
-        print(f"Executing bidi stream with initial_state: {initial_state}")
         state = initial_state
 
         async def attempt():
-            print("New attempt for bidi stream operation.")
             requests_generator = self._strategy.generate_requests(state)
             stream = self._send_and_recv(requests_generator, state)
             try:
@@ -61,7 +58,7 @@ async def attempt():
                 return
             except Exception as e:
                 if retry_policy._predicate(e):
-                    print(
+                    logger.info(
                         f"Bidi stream operation failed: {e}. Attempting state recovery and retry."
                     )
                     await self._strategy.recover_state_on_failure(e, state)
diff --git a/google/cloud/storage/_experimental/asyncio/retry/writes_resumption_strategy.py b/google/cloud/storage/_experimental/asyncio/retry/writes_resumption_strategy.py
@@ -49,9 +49,6 @@ def __init__(
         user_buffer: IO[bytes],
         flush_interval: Optional[int] = None,
     ):
-        print(
-            f"Initializing _WriteState with chunk_size: {chunk_size}, flush_interval: {flush_interval}"
-        )
         self.chunk_size = chunk_size
         self.user_buffer = user_buffer
         self.total_size = self.user_buffer.getbuffer().nbytes
@@ -89,7 +86,6 @@ def generate_requests(self, state: Dict[str, Any]):
         to ensure the server persists the final data and returns the updated state.
         """
         write_state: _WriteState = state["write_state"]
-        print(f"Generating requests with state: {write_state}")
 
         # If this is a retry/redirect, yield a state lookup request first
         # This allows the sender to get current persisted_size before proceeding
@@ -98,14 +94,12 @@ def generate_requests(self, state: Dict[str, Any]):
             or write_state.bytes_sent > write_state.persisted_size
         ):
             # Yield an open/state-lookup request with no data
-            print("Yielding state_lookup request.")
             yield storage_type.BidiWriteObjectRequest(state_lookup=True)
 
         # The buffer should already be seeked to the correct position (persisted_size)
         # by the `recover_state_on_failure` method before this is called.
         while not write_state.is_finalized:
             chunk = write_state.user_buffer.read(write_state.chunk_size)
-            print(f"Read chunk of size: {len(chunk)}")
 
             if not chunk:
                 break
@@ -126,7 +120,6 @@ def generate_requests(self, state: Dict[str, Any]):
 
             write_state.bytes_sent += chunk_len
             write_state.bytes_since_last_flush += chunk_len
-            print(f"Yielding request with offset: {request.write_offset}")
 
             is_flush_point = (
                 write_state.flush_interval
@@ -137,11 +130,9 @@ def generate_requests(self, state: Dict[str, Any]):
                 request.flush = True
                 request.state_lookup = True
                 write_state.bytes_since_last_flush = 0
-                print("Marking request with flush=True and state_lookup=True")
             elif is_flush_point:
                 request.flush = True
                 write_state.bytes_since_last_flush = 0
-                print("Marking request with flush=True")
 
             yield request
 
@@ -150,7 +141,6 @@ def update_state_from_response(
     ) -> None:
         """Processes a server response and updates the write state."""
         write_state: _WriteState = state["write_state"]
-        print(f"Updating state from response: {response}")
         if response is None:
             return
         if response.persisted_size:
@@ -163,7 +153,6 @@ def update_state_from_response(
             write_state.persisted_size = response.resource.size
             if response.resource.finalize_time:
                 write_state.is_finalized = True
-        print(f"New state: {write_state}")
 
     async def recover_state_on_failure(
         self, error: Exception, state: Dict[str, Any]
@@ -174,7 +163,6 @@ async def recover_state_on_failure(
         This method rewinds the user buffer and internal byte tracking to the
         last confirmed 'persisted_size' from the server.
         """
-        print(f"Recovering from error: {error}")
         write_state: _WriteState = state["write_state"]
 
         redirect_proto = None
@@ -193,9 +181,7 @@ async def recover_state_on_failure(
 
         # We must assume any data sent beyond 'persisted_size' was lost.
         # Reset the user buffer to the last known good byte confirmed by the server.
-        print(f"Seeking buffer to: {write_state.persisted_size}")
         write_state.user_buffer.seek(write_state.persisted_size)
         write_state.bytes_sent = write_state.persisted_size
         write_state.bytes_since_last_flush = 0
-        print(f"Recovered state: {write_state}")
 
