googleapis
diff --git a/‎packages/google-cloud-bigquery-storage/google/cloud/bigquery_storage_v1/gapic_types.py‎
Lines changed: 2 additions & 1 deletion b/‎packages/google-cloud-bigquery-storage/google/cloud/bigquery_storage_v1/gapic_types.py‎
Lines changed: 2 additions & 1 deletion
diff --git a/‎packages/google-cloud-bigquery-storage/google/cloud/bigquery_storage_v1/reader.py‎
Lines changed: 7 additions & 9 deletions b/‎packages/google-cloud-bigquery-storage/google/cloud/bigquery_storage_v1/reader.py‎
Lines changed: 7 additions & 9 deletions
diff --git a/‎packages/google-cloud-bigquery-storage/google/cloud/bigquery_storage_v1/writer.py‎
Lines changed: 17 additions & 14 deletions b/‎packages/google-cloud-bigquery-storage/google/cloud/bigquery_storage_v1/writer.py‎
Lines changed: 17 additions & 14 deletions
diff --git a/‎packages/google-cloud-bigquery-storage/google/cloud/bigquery_storage_v1beta2/writer.py‎
Lines changed: 13 additions & 10 deletions b/‎packages/google-cloud-bigquery-storage/google/cloud/bigquery_storage_v1beta2/writer.py‎
Lines changed: 13 additions & 10 deletions
diff --git a/‎packages/google-cloud-bigquery-storage/noxfile.py‎
Lines changed: 2 additions & 1 deletion b/‎packages/google-cloud-bigquery-storage/noxfile.py‎
Lines changed: 2 additions & 1 deletion
diff --git a/‎packages/google-cloud-bigquery-storage/tests/unit/test_writer_v1.py‎
Lines changed: 2 additions & 2 deletions b/‎packages/google-cloud-bigquery-storage/tests/unit/test_writer_v1.py‎
Lines changed: 2 additions & 2 deletions
@@ -22,7 +22,8 @@
 
 fromgoogle.protobufimportmessageasprotobuf_message
 fromgoogle.protobufimporttimestamp_pb2
-importproto
+
+importproto# type: ignore
 
 fromgoogle.cloud.bigquery_storage_v1.typesimportarrow, avro, storage, stream
 
 
@@ -22,23 +22,21 @@
 try:
 importfastavro
 exceptImportError: # pragma: NO COVER
-fastavro=None
+fastavro=None# type: ignore
 importgoogle.api_core.exceptions
-importgoogle.rpc.error_details_pb2
+importgoogle.rpc.error_details_pb2# type: ignore
 
 try:
 importpandas
 exceptImportError: # pragma: NO COVER
-pandas=None
-try:
-importpyarrow
-exceptImportError: # pragma: NO COVER
-pyarrow=None
+pandas=None# type: ignore
 
 try:
-importpyarrow
+# TODO(https://github.com/apache/arrow/issues/32609):
+# Remove `type: ignore` once this bug is fixed
+importpyarrow# type: ignore
 exceptImportError: # pragma: NO COVER
-pyarrow=None
+pyarrow=None# type: ignore
 
 
 _STREAM_RESUMPTION_EXCEPTIONS= (
 
@@ -20,12 +20,12 @@
 importqueue
 importthreading
 importtime
-fromtypingimportCallable, Optional, Sequence, Tuple
+fromtypingimportCallable, List, Optional, Sequence, Tuple, Union
 
 fromgoogle.api_coreimportbidi, exceptions
 fromgoogle.api_core.futureimportpollingaspolling_future
 importgoogle.api_core.retry
-importgrpc
+importgrpc# type: ignore
 
 fromgoogle.cloud.bigquery_storage_v1importexceptionsasbqstorage_exceptions
 fromgoogle.cloud.bigquery_storage_v1importgapic_versionaspackage_version
@@ -44,7 +44,7 @@
 _DEFAULT_TIMEOUT=600
 
 
-def_wrap_as_exception(maybe_exception) ->Exception:
+def_wrap_as_exception(maybe_exception) ->BaseException:
 """Wrap an object as a Python exception, if needed.
  Args:
  maybe_exception (Any): The object to wrap, usually a gRPC exception class.
@@ -109,12 +109,12 @@ def __init__(
  """
 self._client=client
 self._closed=False
-self._close_callbacks= []
+self._close_callbacks: List[Callable]= []
 self._metadata=metadata
 self._thread_lock=threading.RLock()
-self._closed_connection= {}
+self._closed_connection: Union[_Connection|None] =None
 
-self._stream_name=None
+self._stream_name: str=""
 
 # Make a deepcopy of the template and clear the proto3-only fields
 self._initial_request_template=_process_request_template(
@@ -156,7 +156,7 @@ def send(self, request: gapic_types.AppendRowsRequest) -> AppendRowsFuture:
  A future, which can be used to process the response when it
  arrives.
  """
-ifself._stream_nameisNone:
+ifnotself._stream_name:
 self._stream_name=request.write_stream
 elifrequest.write_stream!=self._stream_name:
 raiseValueError(
@@ -211,7 +211,7 @@ def _renew_connection(self, reason: Optional[Exception] = None) -> None:
 # critical section, this step is not guaranteed to be atomic.
 _closed_connection._shutdown(reason=reason)
 
-def_on_rpc_done(self, reason: Optional[Exception] =None) ->None:
+def_on_rpc_done(self, reason: Optional[BaseException] =None) ->None:
 """Callback passecd to _Connection. It's called when the RPC connection
  is closed without recovery. Spins up a new thread to call the helper
  function `_renew_connection()`, which creates a new connection and
@@ -254,10 +254,10 @@ def __init__(
 self._metadata=metadata
 self._thread_lock=threading.RLock()
 
-self._rpc=None
-self._consumer=None
-self._stream_name=None
-self._queue=queue.Queue()
+self._rpc: Union[bidi.BidiRpc|None]=None
+self._consumer: Union[bidi.BackgroundConsumer|None]=None
+self._stream_name: str=""
+self._queue: queue.Queue[AppendRowsFuture]=queue.Queue()
 
 # statuses
 self._closed=False
@@ -429,7 +429,8 @@ def send(self, request: gapic_types.AppendRowsRequest) -> AppendRowsFuture:
 # pull it off and notify completion.
 future=AppendRowsFuture(self._writer)
 self._queue.put(future)
-self._rpc.send(request)
+ifself._rpcisnotNone:
+self._rpc.send(request)
 returnfuture
 
 def_shutdown(self, reason: Optional[Exception] =None) ->None:
@@ -447,7 +448,8 @@ def _shutdown(self, reason: Optional[Exception] = None) -> None:
 # Stop consuming messages.
 ifself.is_active:
 _LOGGER.debug("Stopping consumer.")
-self._consumer.stop()
+ifself._consumerisnotNone:
+self._consumer.stop()
 self._consumer=None
 
 ifself._rpcisnotNone:
@@ -462,6 +464,7 @@ def _shutdown(self, reason: Optional[Exception] = None) -> None:
 # stopped (or at least is attempting to stop), we won't get
 # response callbacks to populate the remaining futures.
 future=self._queue.get_nowait()
+exc: Union[Exception, bqstorage_exceptions.StreamClosedError]
 ifreasonisNone:
 exc=bqstorage_exceptions.StreamClosedError(
 "Stream closed before receiving a response."
 
@@ -20,12 +20,12 @@
 importqueue
 importthreading
 importtime
-fromtypingimportCallable, Optional, Sequence, Tuple
+fromtypingimportCallable, List, Optional, Sequence, Tuple, Union
 
 fromgoogle.api_coreimportbidi, exceptions
 fromgoogle.api_core.futureimportpollingaspolling_future
 importgoogle.api_core.retry
-importgrpc
+importgrpc# type: ignore
 
 fromgoogle.cloud.bigquery_storage_v1beta2importexceptionsasbqstorage_exceptions
 fromgoogle.cloud.bigquery_storage_v1beta2importtypesasgapic_types
@@ -43,7 +43,7 @@
 _DEFAULT_TIMEOUT=600
 
 
-def_wrap_as_exception(maybe_exception) ->Exception:
+def_wrap_as_exception(maybe_exception) ->Union[BaseException]:
 """Wrap an object as a Python exception, if needed.
  Args:
  maybe_exception (Any): The object to wrap, usually a gRPC exception class.
@@ -85,19 +85,19 @@ def __init__(
 self._client=client
 self._closing=threading.Lock()
 self._closed=False
-self._close_callbacks= []
-self._futures_queue=queue.Queue()
+self._close_callbacks: List[Callable]= []
+self._futures_queue: queue.Queue[AppendRowsFuture]=queue.Queue()
 self._inital_request_template=initial_request_template
 self._metadata=metadata
 
 # Only one call to `send()` should attempt to open the RPC.
 self._opening=threading.Lock()
 
-self._rpc=None
-self._stream_name=None
+self._rpc: Union[bidi.BidiRpc|None]=None
+self._stream_name: str=""
 
 # The threads created in ``._open()``.
-self._consumer=None
+self._consumer: Union[bidi.BackgroundConsumer|None]=None
 
 @property
 defis_active(self) ->bool:
@@ -251,7 +251,8 @@ def send(self, request: gapic_types.AppendRowsRequest) -> "AppendRowsFuture":
 # pull it off and notify completion.
 future=AppendRowsFuture(self)
 self._futures_queue.put(future)
-self._rpc.send(request)
+ifself._rpcisnotNone:
+self._rpc.send(request)
 returnfuture
 
 def_on_response(self, response: gapic_types.AppendRowsResponse):
@@ -302,7 +303,8 @@ def _shutdown(self, reason: Optional[Exception] = None):
 # Stop consuming messages.
 ifself.is_active:
 _LOGGER.debug("Stopping consumer.")
-self._consumer.stop()
+ifself._consumerisnotNone:
+self._consumer.stop()
 self._consumer=None
 
 ifself._rpcisnotNone:
@@ -318,6 +320,7 @@ def _shutdown(self, reason: Optional[Exception] = None):
 # stopped (or at least is attempting to stop), we won't get
 # response callbacks to populate the remaining futures.
 future=self._futures_queue.get_nowait()
+exc: Union[Exception, bqstorage_exceptions.StreamClosedError]
 ifreasonisNone:
 exc=bqstorage_exceptions.StreamClosedError(
 "Stream closed before receiving a response."
 
@@ -112,8 +112,9 @@ def mypy(session):
 "mypy<1.16.0",
 "types-requests",
 "types-protobuf",
+"pandas-stubs",
  )
-session.install(".")
+session.install(".[fastavro]")
 session.run(
 "mypy",
 "-p",
 
@@ -66,7 +66,7 @@ def test_ctor_defaults(self):
 assertstream._closedisFalse
 assertnotstream._close_callbacks
 assertstream._metadata== ()
-assertstream._stream_nameisNone
+assertstream._stream_name==""
 assertisinstance(stream._thread_lock, type(threading.RLock()))
 
 assertisinstance(stream._connection, _Connection)
@@ -229,7 +229,7 @@ def test_ctor_defaults(self):
 assertisinstance(connection._thread_lock, type(threading.RLock()))
 assertconnection._rpcisNone
 assertconnection._consumerisNone
-assertconnection._stream_nameisNone
+assertconnection._stream_name==""
 assertisinstance(connection._queue, queue.Queue)
 
 assertconnection._closedisFalse