fix: handle AttributeError in bigquery_storage writer

[parthea]

This PR fixes an issue where an AttributeError could be raised here if either self._rpc or the self._consumer is null. @tswast and I saw this AttributeError locally with the changes from googleapis/python-api-core#357

See stack trace below

Exception has occurred: AttributeError
'NoneType' object has no attribute 'is_active'
  File "[/usr/local/google/home/partheniou/git/python-bigquery-storage/google/cloud/bigquery_storage_v1/writer.py]()", line 192, in _open
    while not self._rpc.is_active and self._consumer.is_active:
  File "[/usr/local/google/home/partheniou/git/python-bigquery-storage/google/cloud/bigquery_storage_v1/writer.py]()", line 230, in send
    return self._open(request)
  File "[/usr/local/google/home/partheniou/git/python-bigquery-storage/samples/snippets/append_rows_proto2.py]()", line 123, in append_rows_proto2
    response_future_1 = append_rows_stream.send(request)
  File "[/usr/local/google/home/partheniou/git/python-bigquery-storage/samples/snippets/append_rows_proto2_test.py]()", line 58, in test_append_rows_proto2
    append_rows_proto2.append_rows_proto2(

[tswast]

Looks good! Would be great if we could get a unit test or two checking this behavior (closed during _open)

[parthea]

I'm only able to get attribute error when I change the client code here to the code in googleapis/python-api-core#268 (comment). I suggest proceeding without a test as I'm not able to get the code to fail without changing the client code in writer.py.

[parthea]

Here is an example of the failure locally when using the modified version of writer.py from googleapis/python-api-core#268 (comment)

______________________________________________________________________________________________________________ test_append_rows_proto2[non-US] _______________________________________________________________________________________________________________

capsys = <_pytest.capture.CaptureFixture object at 0x7f56ca07ff70>, bigquery_client = <google.cloud.bigquery.client.Client object at 0x7f56ca1026d0>
sample_data_table = 'python-docs-samples-tests.python_bigquery_storage_samples_snippets_20220613232605_aa4ab3.append_rows_proto2_1390'

    def test_append_rows_proto2(
        capsys: pytest.CaptureFixture,
        bigquery_client: bigquery.Client,
        sample_data_table: str,
    ):
        project_id, dataset_id, table_id = sample_data_table.split(".")
>       append_rows_proto2.append_rows_proto2(
            project_id=project_id, dataset_id=dataset_id, table_id=table_id
        )

samples/snippets/append_rows_proto2_test.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
samples/snippets/append_rows_proto2.py:124: in append_rows_proto2
    response_future_1 = append_rows_stream.send(request)
google/cloud/bigquery_storage_v1/writer.py:221: in send
    return self._open(request)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <google.cloud.bigquery_storage_v1.writer.AppendRowsStream object at 0x7f56e1859af0>
initial_request = offset {
}
proto_rows {
  rows {
    serialized_rows: "\010\001\022\rHello, World!\031\000\000\000\000\000\000\360\177...1\004"
    serialized_rows: " \270\213\004\200\001\005"
    serialized_rows: "*\020Auf Wiedersehen!\200\001\006"
  }
}

timeout = 600

    def _open(
        self,
        initial_request: gapic_types.AppendRowsRequest,
        timeout: float = _DEFAULT_TIMEOUT,
    ) -> "AppendRowsFuture":
        """Open an append rows stream.
    
        This is automatically called by the first call to the
        :attr:`google.cloud.bigquery_storage_v1.writer.AppendRowsStream.send`
        method.
    
        Args:
            initial_request:
                The initial request to start the stream. Must have
                :attr:`google.cloud.bigquery_storage_v1.types.AppendRowsRequest.write_stream`
                and ``proto_rows.writer_schema.proto_descriptor`` and
                properties populated.
            timeout:
                How long (in seconds) to wait for the stream to be ready.
    
        Returns:
            A future, which can be used to process the response to the initial
            request when it arrives.
        """
        if self.is_active:
            raise ValueError("This manager is already open.")
    
        if self._closed:
            raise bqstorage_exceptions.StreamClosedError(
                "This manager has been closed and can not be re-used."
            )
    
        start_time = time.monotonic()
        self._inital_request = initial_request
        self._stream_name = initial_request.write_stream
    
        inital_response_future = AppendRowsFuture(self)
        self._futures_queue.put(inital_response_future)
    
        self._rpc = bidi.BidiRpc(
            self._client.append_rows,
            initial_request=self._inital_request,
            # TODO: pass in retry and timeout. Blocked by
            # https://github.com/googleapis/python-api-core/issues/262
            metadata=tuple(
                itertools.chain(
                    self._metadata,
                    # This header is required so that the BigQuery Storage API
                    # knows which region to route the request to.
                    (("x-goog-request-params", f"write_stream={self._stream_name}"),),
                )
            ),
        )
        self._rpc.add_done_callback(self._on_rpc_done)
    
        self._consumer = bidi.BackgroundConsumer(self._rpc, self._on_response)
        self._consumer.start()
    
        # Make sure RPC has started before returning.
        # Without this, consumers may get:
        #
        # ValueError: Can not send() on an RPC that has never been open()ed.
        #
        # when they try to send a request.
>       while not self._rpc.is_active and self._consumer.is_active:
E       AttributeError: 'NoneType' object has no attribute 'is_active'

google/cloud/bigquery_storage_v1/writer.py:183: AttributeError

Dismiss stale review