fix(bigquery): handle not found in teardown and harden socket cleanup

Author: shuoweil

packages/google-cloud-bigquery/google/cloud/bigquery/dbapi/connection.py

@@ -78,10 +78,14 @@ def close(self):
         """
         self._closed = True
 
-        if self._owns_client:
+        for cursor_ in list(self._cursors_created):
+            if not cursor_._closed:
+                cursor_.close()
+
+        if self._owns_client and self._client is not None:
             self._client.close()
 
-        if self._owns_bqstorage_client:
+        if self._owns_bqstorage_client and self._bqstorage_client is not None:
             # There is no close() on the BQ Storage client itself.
             transport = self._bqstorage_client.transport
             transport.close()
@@ -95,10 +99,6 @@ def close(self):
             if channel is not None:
                 channel.close()
 
-        for cursor_ in self._cursors_created:
-            if not cursor_._closed:
-                cursor_.close()
-
     def commit(self):
         """No-op, but for consistency raise an error if connection is closed."""
 

packages/google-cloud-bigquery/tests/system/test_client.py

@@ -204,12 +204,16 @@ def _still_in_use(bad_request):
             tag_key = key_values.pop()
 
             # Delete tag values first
-            [
-                tag_values_client.delete_tag_value(name=tag_value.name).result()
-                for tag_value in key_values
-            ]
+            for tag_value in key_values:
+                try:
+                    tag_values_client.delete_tag_value(name=tag_value.name).result()
+                except NotFound:
+                    pass
 
-            tag_keys_client.delete_tag_key(name=tag_key.name).result()
+            try:
+                tag_keys_client.delete_tag_key(name=tag_key.name).result()
+            except NotFound:
+                pass
 
     def test_get_service_account_email(self):
         client = Config.CLIENT
@@ -2203,11 +2207,15 @@ def test_dbapi_connection_does_not_leak_sockets(self):
             rows = cursor.fetchall()
             self.assertEqual(len(rows), 100000)
 
+            cursor.close()
             connection.close()
+
+        del connection, cursor, rows
         import gc
 
         gc.collect()
         for _ in range(60):  # Wait up to 6 seconds for background socket cleanup
+            gc.collect()
             conn_end = current_process.net_connections()
             conn_count_end = len(conn_end)
             new_conns_remaining = [