Add SchemaField serialization and deserialization.

bigquery/google/cloud/bigquery/schema.py

@@ -43,6 +43,25 @@ def __init__(self, name, field_type, mode='NULLABLE',
         self._description = description
         self._fields = tuple(fields)
 
+    @classmethod
+    def from_api_repr(cls, api_repr):
+        """Return a ``SchemaField`` object deserialized from a dictionary.
+
+        Args:
+            api_repr (Mapping[str, str]): The serialized representation
+                of the SchemaField, such as what is output by
+                :meth:`to_api_repr`.
+
+        Returns:
+            SchemaField: The ``SchemaField`` object.
+        """
+        return cls(
+            field_type=api_repr['type'].upper(),
+            fields=[cls.from_api_repr(f) for f in api_repr.get('fields', ())],
+            mode=api_repr['mode'].upper(),
+            name=api_repr['name'],
+        )
+
     @property
     def name(self):
         """str: The name of the field."""
@@ -84,6 +103,28 @@ def fields(self):
         """
         return self._fields
 
+    def to_api_repr(self):
+        """Return a dictionary representing this schema field.
+
+        Returns:
+            dict: A dictionary representing the SchemaField in a serialized
+                form.
+        """
+        # Put together the basic representation. See http://bit.ly/2hOAT5u.
+        answer = {
+            'mode': self.mode.lower(),
+            'name': self.name,
+            'type': self.field_type.lower(),
+        }
+
+        # If this is a RECORD type, then sub-fields are also included,
+        # add this to the serialized representation.
+        if self.field_type.upper() == 'RECORD':
+            answer['fields'] = [f.to_api_repr() for f in self.fields]
+
+        # Done; return the serialized dictionary.
+        return answer
+
     def _key(self):
         """A tuple key that unique-ly describes this field.
 

bigquery/tests/unit/test_schema.py

@@ -61,6 +61,47 @@ def test_constructor_subfields(self):
         self.assertIs(field._fields[0], sub_field1)
         self.assertIs(field._fields[1], sub_field2)
 
+    def test_to_api_repr(self):
+        field = self._make_one('foo', 'INTEGER', 'NULLABLE')
+        self.assertEqual(field.to_api_repr(), {
+            'mode': 'nullable',
+            'name': 'foo',
+            'type': 'integer',
+        })
+
+    def test_to_api_repr_with_subfield(self):
+        subfield = self._make_one('bar', 'INTEGER', 'NULLABLE')
+        field = self._make_one('foo', 'RECORD', 'REQUIRED', fields=(subfield,))
+        self.assertEqual(field.to_api_repr(), {
+            'fields': [{
+                'mode': 'nullable',
+                'name': 'bar',
+                'type': 'integer',
+            }],
+            'mode': 'required',
+            'name': 'foo',
+            'type': 'record',
+        })
+
+    def test_from_api_repr(self):
+        field = self._get_target_class().from_api_repr({
+            'fields': [{
+                'mode': 'nullable',
+                'name': 'bar',
+                'type': 'integer',
+            }],
+            'mode': 'required',
+            'name': 'foo',
+            'type': 'record',
+        })
+        self.assertEqual(field.name, 'foo')
+        self.assertEqual(field.field_type, 'RECORD')
+        self.assertEqual(field.mode, 'REQUIRED')
+        self.assertEqual(len(field.fields), 1)
+        self.assertEqual(field.fields[0].name, 'bar')
+        self.assertEqual(field.fields[0].field_type, 'INTEGER')
+        self.assertEqual(field.fields[0].mode, 'NULLABLE')
+
     def test_name_property(self):
         name = 'lemon-ness'
         schema_field = self._make_one(name, 'INTEGER')