bigquery: add streaming buffer info (#4161)

jba · web-flow · commit c8521b81da74 · 2017-10-11T20:07:22.000-04:00
Unfortunately there's no good way to write a system test for this,
since you can never be sure that one gets created. But I informally
verified that the code works by running create_rows a lot until
I got a streaming buffer.
diff --git a/bigquery/google/cloud/bigquery/table.py b/bigquery/google/cloud/bigquery/table.py
@@ -524,6 +524,12 @@ def view_use_legacy_sql(self, value):
             self._properties['view'] = {}
         self._properties['view']['useLegacySql'] = value
 
+    @property
+    def streaming_buffer(self):
+        sb = self._properties.get('streamingBuffer')
+        if sb is not None:
+            return StreamingBuffer(sb)
+
     @classmethod
     def from_api_repr(cls, resource):
         """Factory:  construct a table given its API representation
@@ -658,6 +664,23 @@ def _row_from_mapping(mapping, schema):
     return tuple(row)
 
 
+class StreamingBuffer(object):
+    """Information about a table's streaming buffer.
+
+    See https://cloud.google.com/bigquery/streaming-data-into-bigquery.
+
+    :type resource: dict
+    :param resource: streaming buffer representation returned from the API
+    """
+
+    def __init__(self, resource):
+        self.estimated_bytes = int(resource['estimatedBytes'])
+        self.estimated_rows = int(resource['estimatedRows'])
+        # time is in milliseconds since the epoch.
+        self.oldest_entry_time = _datetime_from_microseconds(
+            1000.0 * int(resource['oldestEntryTime']))
+
+
 def _parse_schema_resource(info):
     """Parse a resource fragment into a schema field.
 
diff --git a/bigquery/tests/unit/test_table.py b/bigquery/tests/unit/test_table.py
@@ -174,6 +174,8 @@ def _setUpConstants(self):
         self.RESOURCE_URL = 'http://example.com/path/to/resource'
         self.NUM_BYTES = 12345
         self.NUM_ROWS = 67
+        self.NUM_EST_BYTES = 1234
+        self.NUM_EST_ROWS = 23
 
     def _makeResource(self):
         self._setUpConstants()
@@ -194,6 +196,10 @@ def _makeResource(self):
             'numRows': self.NUM_ROWS,
             'numBytes': self.NUM_BYTES,
             'type': 'TABLE',
+            'streamingBuffer': {
+                'estimatedRows': str(self.NUM_EST_ROWS),
+                'estimatedBytes': str(self.NUM_EST_BYTES),
+                'oldestEntryTime': self.WHEN_TS * 1000},
         }
 
     def _verifyReadonlyResourceProperties(self, table, resource):
@@ -222,6 +228,16 @@ def _verifyReadonlyResourceProperties(self, table, resource):
         else:
             self.assertIsNone(table.self_link)
 
+        if 'streamingBuffer' in resource:
+            self.assertEqual(table.streaming_buffer.estimated_rows,
+                             self.NUM_EST_ROWS)
+            self.assertEqual(table.streaming_buffer.estimated_bytes,
+                             self.NUM_EST_BYTES)
+            self.assertEqual(table.streaming_buffer.oldest_entry_time,
+                             self.WHEN)
+        else:
+            self.assertIsNone(table.streaming_buffer)
+
         self.assertEqual(table.full_table_id, self.TABLE_FULL_ID)
         self.assertEqual(table.table_type,
                          'TABLE' if 'view' not in resource else 'VIEW')
