Phase 1 for storing schemas for later use.

bigquery/google/cloud/bigquery/client.py

@@ -23,6 +23,8 @@
 
 import functools
 import gzip
+import io
+import json
 import os
 import tempfile
 import uuid
@@ -50,6 +52,7 @@
 from google.cloud.bigquery.model import ModelReference
 from google.cloud.bigquery.query import _QueryResults
 from google.cloud.bigquery.retry import DEFAULT_RETRY
+from google.cloud.bigquery.schema import SchemaField
 from google.cloud.bigquery.table import _table_arg_to_table
 from google.cloud.bigquery.table import _table_arg_to_table_ref
 from google.cloud.bigquery.table import Table
@@ -71,6 +74,9 @@
 _READ_LESS_THAN_SIZE = (
     "Size {:d} was specified but the file-like object only had " "{:d} bytes remaining."
 )
+_NEED_JSON_FILE_ARGUMENT = (
+    "The JSON file argument should be a file object or a file path"
+)
 _NEED_TABLE_ARGUMENT = (
     "The table argument should be a table ID string, Table, or TableReference"
 )
@@ -1929,6 +1935,61 @@ def list_rows(
         )
         return row_iterator
 
+    def _schema_from_json_file_object(self, file):
+        """Helper function for schema_from_json that takes a
+           file object that describes a table schema.
+
+           Returns:
+                List of schema field objects.
+        """
+        schema_field_list = list()
+        json_data = json.load(file)
+
+        for field in json_data:
+            schema_field = SchemaField.from_api_repr(field)
+            schema_field_list.append(schema_field)
+
+        return schema_field_list
+
+    def schema_from_json(self, file_or_path):
+        """Takes a file object or file path that contains json that describes
+            a table schema.
+
+            Returns:
+                List of schema field objects.
+        """
+        if isinstance(file_or_path, io.IOBase):
+            return self._schema_from_json_file_object(file_or_path)
+        else:
+            try:
+                with open(file_or_path) as file:
+                    return self._schema_from_json_file_object(file)
+            except OSError:
+                raise ValueError(_NEED_JSON_FILE_ARGUMENT)
+
+    def schema_to_json(self, schema_list, destination):
+        """Takes a list of schema field objects.
+
+            Serializes the list of schema field objects as json to a file.
+
+            Destination is a file path or a file object.
+        """
+        file_obj = None
+        json_schema_list = list()
+
+        for field in schema_list:
+            schema_field = field.to_api_repr()
+            json_schema_list.append(schema_field)
+
+        if isinstance(destination, io.IOBase):
+            destination.write(json.dumps(json_schema_list, indent=2, sort_keys=True))
+        else:
+            try:
+                with open(destination, mode="w") as file:
+                    file.write(json.dumps(json_schema_list, indent=2, sort_keys=True))
+            except OSError:
+                raise ValueError(_NEED_JSON_FILE_ARGUMENT)
+
 
 # pylint: disable=unused-argument
 def _item_to_project(iterator, resource):

bigquery/tests/unit/test_client.py

@@ -5161,3 +5161,166 @@ def test__do_multipart_upload_wrong_size(self):
 
         with pytest.raises(ValueError):
             client._do_multipart_upload(file_obj, {}, file_obj_len + 1, None)
+
+    def test_schema_from_json_with_file_path(self):
+        from google.cloud.bigquery.schema import SchemaField
+
+        file_content = """[
+          {
+            "description": "quarter",
+            "mode": "REQUIRED",
+            "name": "qtr",
+            "type": "STRING"
+          },
+          {
+            "description": "sales representative",
+            "mode": "NULLABLE",
+            "name": "rep",
+            "type": "STRING"
+          },
+          {
+            "description": "total sales",
+            "mode": "NULLABLE",
+            "name": "sales",
+            "type": "FLOAT"
+          }
+        ]"""
+
+        expected = list()
+        expected.append(SchemaField("qtr", "STRING", "REQUIRED", "quarter"))
+        expected.append(
+            SchemaField("rep", "STRING", "NULLABLE", "sales representative")
+        )
+        expected.append(SchemaField("sales", "FLOAT", "NULLABLE", "total sales"))
+
+        client = self._make_client()
+        mock_file_path = "/mocked/file.json"
+
+        open_patch = mock.patch(
+            "builtins.open", new=mock.mock_open(read_data=file_content)
+        )
+
+        with open_patch as _mock_file:
+            actual = client.schema_from_json(mock_file_path)
+            _mock_file.assert_called_once_with(mock_file_path)
+            # This assert is to make sure __exit__ is called in the context
+            # manager that opens the file in the function
+            _mock_file().__exit__.assert_called_once_with(None, None, None)
+
+        assert expected == actual
+
+    def test_schema_from_json_with_file_object(self):
+        from google.cloud.bigquery.schema import SchemaField
+
+        file_content = """[
+          {
+            "description": "quarter",
+            "mode": "REQUIRED",
+            "name": "qtr",
+            "type": "STRING"
+          },
+          {
+            "description": "sales representative",
+            "mode": "NULLABLE",
+            "name": "rep",
+            "type": "STRING"
+          },
+          {
+            "description": "total sales",
+            "mode": "NULLABLE",
+            "name": "sales",
+            "type": "FLOAT"
+          }
+        ]"""
+
+        expected = list()
+        expected.append(SchemaField("qtr", "STRING", "REQUIRED", "quarter"))
+        expected.append(
+            SchemaField("rep", "STRING", "NULLABLE", "sales representative")
+        )
+        expected.append(SchemaField("sales", "FLOAT", "NULLABLE", "total sales"))
+
+        client = self._make_client()
+
+        fake_file = io.StringIO(file_content)
+        actual = client.schema_from_json(fake_file)
+
+        assert expected == actual
+
+    def test_schema_to_json_with_file_path(self):
+        from google.cloud.bigquery.schema import SchemaField
+
+        file_content = """[
+  {
+    "description": "quarter",
+    "mode": "REQUIRED",
+    "name": "qtr",
+    "type": "STRING"
+  },
+  {
+    "description": "sales representative",
+    "mode": "NULLABLE",
+    "name": "rep",
+    "type": "STRING"
+  },
+  {
+    "description": "total sales",
+    "mode": "NULLABLE",
+    "name": "sales",
+    "type": "FLOAT"
+  }
+]"""
+        schema_list = list()
+        schema_list.append(SchemaField("qtr", "STRING", "REQUIRED", "quarter"))
+        schema_list.append(
+            SchemaField("rep", "STRING", "NULLABLE", "sales representative")
+        )
+        schema_list.append(SchemaField("sales", "FLOAT", "NULLABLE", "total sales"))
+
+        client = self._make_client()
+        mock_file_path = "/mocked/file.json"
+
+        open_patch = mock.patch("builtins.open", mock.mock_open())
+        with open_patch as _mock_file:
+            actual = client.schema_to_json(schema_list, mock_file_path)
+            _mock_file.assert_called_once_with(mock_file_path, mode="w")
+            _mock_file().write.assert_called_once_with(file_content)
+            # This assert is to make sure __exit__ is called in the context
+            # manager that opens the file in the function
+            _mock_file().__exit__.assert_called_once_with(None, None, None)
+
+    def test_schema_to_json_with_file_object(self):
+        from google.cloud.bigquery.schema import SchemaField
+
+        file_content = """[
+  {
+    "description": "quarter",
+    "mode": "REQUIRED",
+    "name": "qtr",
+    "type": "STRING"
+  },
+  {
+    "description": "sales representative",
+    "mode": "NULLABLE",
+    "name": "rep",
+    "type": "STRING"
+  },
+  {
+    "description": "total sales",
+    "mode": "NULLABLE",
+    "name": "sales",
+    "type": "FLOAT"
+  }
+]"""
+        schema_list = list()
+        schema_list.append(SchemaField("qtr", "STRING", "REQUIRED", "quarter"))
+        schema_list.append(
+            SchemaField("rep", "STRING", "NULLABLE", "sales representative")
+        )
+        schema_list.append(SchemaField("sales", "FLOAT", "NULLABLE", "total sales"))
+
+        fake_file = io.StringIO()
+        client = self._make_client()
+
+        client.schema_to_json(schema_list, fake_file)
+        assert file_content == fake_file.getvalue()