feat: adds Schema class and modifies schema handling

google/cloud/bigquery/client.py

@@ -3680,7 +3680,7 @@ def insert_rows(
         if selected_fields is not None:
             schema = selected_fields
 
-        if len(schema) == 0:
+        if not schema:
             raise ValueError(
                 (
                     "Could not determine schema for table '{}'. Call client.get_table() "
@@ -4029,7 +4029,7 @@ def list_rows(
 
         # No schema, but no selected_fields. Assume the developer wants all
         # columns, so get the table resource for them rather than failing.
-        elif len(schema) == 0:
+        elif not schema:
             table = self.get_table(table.reference, retry=retry, timeout=timeout)
             schema = table.schema
 

google/cloud/bigquery/schema.py

@@ -549,33 +549,29 @@ def _build_schema_resource(fields):
 
 
 def _to_schema_fields(schema):
-    """Coerce `schema` to a list of schema field instances.
-
-    Args:
-        schema(Sequence[Union[ \
-            :class:`~google.cloud.bigquery.schema.SchemaField`, \
-            Mapping[str, Any] \
-        ]]):
-            Table schema to convert. If some items are passed as mappings,
-            their content must be compatible with
-            :meth:`~google.cloud.bigquery.schema.SchemaField.from_api_repr`.
-
-    Returns:
-        Sequence[:class:`~google.cloud.bigquery.schema.SchemaField`]
-
-    Raises:
-        Exception: If ``schema`` is not a sequence, or if any item in the
-        sequence is not a :class:`~google.cloud.bigquery.schema.SchemaField`
-        instance or a compatible mapping representation of the field.
+    """TODO docstring
+    CAST a list of elements to either:
+    * a Schema object with SchemaFields and an attribute
+    * a list of SchemaFields but no attribute
     """
-
     for field in schema:
         if not isinstance(field, (SchemaField, collections.abc.Mapping)):
             raise ValueError(
                 "Schema items must either be fields or compatible "
                 "mapping representations."
             )
 
+    if isinstance(schema, Schema):
+        schema = Schema(
+            [
+                field
+                if isinstance(field, SchemaField)
+                else SchemaField.from_api_repr(field)
+                for field in schema
+            ],
+            foreign_type_info=schema.foreign_type_info,
+        )
+        return schema
     return [
         field if isinstance(field, SchemaField) else SchemaField.from_api_repr(field)
         for field in schema
@@ -796,8 +792,6 @@ def serde_info(self) -> Any:
         prop = _get_sub_prop(self._properties, ["serDeInfo"])
         if prop is not None:
             prop = StorageDescriptor().from_api_repr(prop)
-            print(f"DINOSAUR prop: {prop}")
-
         return prop
 
     @serde_info.setter
@@ -921,3 +915,83 @@ def from_api_repr(cls, resource: dict) -> SerDeInfo:
         config = cls("")
         config._properties = copy.deepcopy(resource)
         return config
+
+
+class Schema:
+    # TODO docstrings and type hints
+    def __init__(self, fields=None, foreign_type_info=None):
+        self._properties = {}
+        self._fields = [] if fields is None else list(fields)  # Internal List
+        self.foreign_type_info = foreign_type_info
+
+    @property
+    def foreign_type_info(self) -> Any:
+        """TODO: docstring"""
+        return self._properties.get("foreignTypeInfo")
+
+    @foreign_type_info.setter
+    def foreign_type_info(self, value: str) -> None:
+        value = _isinstance_or_raise(value, str, none_allowed=True)
+        self._properties["foreignTypeInfo"] = value
+
+    @property
+    def _fields(self) -> Any:
+        """TODO: docstring"""
+        return self._properties.get("_fields")
+
+    @_fields.setter
+    def _fields(self, value: list) -> None:
+        value = _isinstance_or_raise(value, list, none_allowed=True)
+        self._properties["_fields"] = value
+
+    def __len__(self):
+        return len(self._fields)
+
+    def __getitem__(self, index):
+        return self._fields[index]
+
+    def __setitem__(self, index, value):
+        self._fields[index] = value
+
+    def __delitem__(self, index):
+        del self._fields[index]
+
+    def __iter__(self):
+        return iter(self._fields)
+
+    def __str__(self):
+        return f"Schema({self._fields}, {self.foreign_type_info})"
+
+    def __repr__(self):
+        return f"Schema({self._fields!r}, {self.foreign_type_info!r})"
+
+    def append(self, item):
+        self._fields.append(item)
+
+    def extend(self, iterable):
+        self._fields.extend(iterable)
+
+    def to_api_repr(self) -> dict:
+        """Build an API representation of this object.
+
+        Returns:
+            Dict[str, Any]:
+                A dictionary in the format used by the BigQuery API.
+        """
+        return copy.deepcopy(self._properties)
+
+    @classmethod
+    def from_api_repr(cls, resource: dict) -> Schema:
+        """Factory: constructs an instance of the class (cls)
+        given its API representation.
+
+        Args:
+            resource (Dict[str, Any]):
+                API representation of the object to be instantiated.
+
+        Returns:
+            An instance of the class initialized with data from 'resource'.
+        """
+        config = cls("")
+        config._properties = copy.deepcopy(resource)
+        return config

google/cloud/bigquery/table.py

@@ -70,6 +70,7 @@
 from google.cloud.bigquery.schema import _build_schema_resource
 from google.cloud.bigquery.schema import _parse_schema_resource
 from google.cloud.bigquery.schema import _to_schema_fields
+from google.cloud.bigquery.schema import Schema
 from google.cloud.bigquery.external_config import ExternalCatalogTableOptions
 
 if typing.TYPE_CHECKING:  # pragma: NO COVER
@@ -452,17 +453,20 @@ def schema(self):
                 instance or a compatible mapping representation of the field.
         """
         prop = self._properties.get(self._PROPERTY_TO_API_FIELD["schema"])
-        if not prop:
+        if not prop:  # if empty Schema, empty list, None
             return []
-        else:
-            return _parse_schema_resource(prop)
+        elif isinstance(prop, Schema):
+            return prop
+        return _parse_schema_resource(prop)
 
     @schema.setter
     def schema(self, value):
         api_field = self._PROPERTY_TO_API_FIELD["schema"]
 
         if value is None:
             self._properties[api_field] = None
+        elif isinstance(value, Schema):
+            self._properties[api_field] = value
         else:
             value = _to_schema_fields(value)
             self._properties[api_field] = {"fields": _build_schema_resource(value)}
@@ -1359,7 +1363,8 @@ def _row_from_mapping(mapping, schema):
     Raises:
         ValueError: If schema is empty.
     """
-    if len(schema) == 0:
+
+    if not schema:
         raise ValueError(_TABLE_HAS_NO_SCHEMA)
 
     row = []

tests/unit/test_client.py

@@ -60,6 +60,7 @@
 from google.cloud.bigquery import ParquetOptions
 import google.cloud.bigquery.retry
 from google.cloud.bigquery.retry import DEFAULT_TIMEOUT
+from google.cloud.bigquery.schema import Schema
 import google.cloud.bigquery.table
 
 from test_utils.imports import maybe_fail_import
@@ -2608,7 +2609,8 @@ def test_update_table_w_schema_None(self):
         sent = {"schema": None}
         self.assertEqual(req[1]["data"], sent)
         self.assertEqual(req[1]["path"], "/%s" % path)
-        self.assertEqual(len(updated_table.schema), 0)
+        valid_options = [Schema(), [], None]
+        self.assertIn(updated_table.schema, valid_options)
 
     def test_update_table_delete_property(self):
         from google.cloud.bigquery.table import Table