Feat: Adds foreign_type_info attribute to table class and adds unit tests.

google/cloud/bigquery/table.py

@@ -66,6 +66,7 @@
 from google.cloud.bigquery.encryption_configuration import EncryptionConfiguration
 from google.cloud.bigquery.enums import DefaultPandasDTypes
 from google.cloud.bigquery.external_config import ExternalConfig
+from google.cloud.bigquery import schema as _schema
 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
@@ -411,6 +412,7 @@ class Table(_TableBase):
         "max_staleness": "maxStaleness",
         "resource_tags": "resourceTags",
         "external_catalog_table_options": "externalCatalogTableOptions",
+        "foreign_type_info": "foreignTypeInfo",
     }
 
     def __init__(self, table_ref, schema=None) -> None:
@@ -451,6 +453,15 @@ def schema(self):
                 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.
+
+        NOTE: If you are referencing a schema for an external catalog table such
+        as a Hive table, it will also be necessary to populate the foreign_type_info
+        attribute. This is not necessary if defining the schema for a BigQuery table.
+
+        For details, see:
+        https://cloud.google.com/bigquery/docs/external-tables
+        https://cloud.google.com/bigquery/docs/datasets-intro#external_datasets
+
         """
         prop = self._properties.get(self._PROPERTY_TO_API_FIELD["schema"])
         if not prop:
@@ -1075,6 +1086,41 @@ def external_catalog_table_options(
                 self._PROPERTY_TO_API_FIELD["external_catalog_table_options"]
             ] = value
 
+    @property
+    def foreign_type_info(self) -> Optional[_schema.ForeignTypeInfo]:
+        """Optional. Specifies metadata of the foreign data type definition in
+        field schema (TableFieldSchema.foreign_type_definition).
+
+        Returns:
+            Optional[schema.ForeignTypeInfo]:
+                Foreign type information, or :data:`None` if not set.
+
+        NOTE: foreign_type_info is only required if you are referencing an
+        external catalog such as a Hive table.
+        For details, see:
+        https://cloud.google.com/bigquery/docs/external-tables
+        https://cloud.google.com/bigquery/docs/datasets-intro#external_datasets
+        """
+
+        prop = self._properties.get(self._PROPERTY_TO_API_FIELD["foreign_type_info"])
+        if prop is not None:
+            return _schema.ForeignTypeInfo.from_api_repr(prop)
+        return None
+
+    @foreign_type_info.setter
+    def foreign_type_info(self, value: Union[_schema.ForeignTypeInfo, dict, None]):
+        value = _helpers._isinstance_or_raise(
+            value,
+            (_schema.ForeignTypeInfo, dict),
+            none_allowed=True,
+        )
+        if isinstance(value, _schema.ForeignTypeInfo):
+            self._properties[
+                self._PROPERTY_TO_API_FIELD["foreign_type_info"]
+            ] = value.to_api_repr()
+        else:
+            self._properties[self._PROPERTY_TO_API_FIELD["foreign_type_info"]] = value
+
     @classmethod
     def from_string(cls, full_table_id: str) -> "Table":
         """Construct a table from fully-qualified table ID.

tests/unit/test_table.py

@@ -31,6 +31,7 @@
 from google.cloud.bigquery import _versions_helpers
 from google.cloud.bigquery import exceptions
 from google.cloud.bigquery import external_config
+from google.cloud.bigquery import schema
 from google.cloud.bigquery.table import TableReference
 from google.cloud.bigquery.dataset import DatasetReference
 
@@ -5993,6 +5994,71 @@ def test_external_catalog_table_options_from_api_repr(self):
         assert result == expected
 
 
+class TestForeignTypeInfo:
+    PROJECT = "test-project"
+    DATASET_ID = "test_dataset"
+    TABLE_ID = "coffee_table"
+    DATASET = DatasetReference(PROJECT, DATASET_ID)
+    TABLEREF = DATASET.table(TABLE_ID)
+    FOREIGNTYPEINFO = {
+        "typeSystem": "TYPE_SYSTEM_UNSPECIFIED",
+    }
+
+    from google.cloud.bigquery.schema import ForeignTypeInfo
+
+    @staticmethod
+    def _get_target_class(self):
+        from google.cloud.bigquery.table import Table
+
+        return Table
+
+    def _make_one(self, *args, **kw):
+        return self._get_target_class(self)(*args, **kw)
+
+    def test_foreign_type_info_default_initialization(self):
+        table = self._make_one(self.TABLEREF)
+        assert table.foreign_type_info is None
+
+    def test_foreign_type_info_valid_inputs(self):
+        table = self._make_one(self.TABLEREF)
+
+        table.foreign_type_info = self.ForeignTypeInfo(
+            type_system="TYPE_SYSTEM_UNSPECIFIED",
+        )
+
+        result = table.foreign_type_info.type_system
+        expected = self.FOREIGNTYPEINFO["typeSystem"]
+        assert result == expected
+
+    def test_foreign_type_info_invalid_inputs(self):
+        table = self._make_one(self.TABLEREF)
+
+        # invalid on the whole
+        with pytest.raises(TypeError, match="Pass .*"):
+            table.foreign_type_info = 123
+
+    def test_foreign_type_info_to_api_repr(self):
+        table = self._make_one(self.TABLEREF)
+
+        table.foreign_type_info = self.ForeignTypeInfo(
+            type_system="TYPE_SYSTEM_UNSPECIFIED",
+        )
+
+        result = table.to_api_repr()["foreignTypeInfo"]
+        expected = self.FOREIGNTYPEINFO
+        assert result == expected
+
+    def test_foreign_type_info_from_api_repr(self):
+        table = self._make_one(self.TABLEREF)
+        table.foreign_type_info = self.FOREIGNTYPEINFO
+
+        fti = schema.ForeignTypeInfo.from_api_repr(self.FOREIGNTYPEINFO)
+
+        result = fti.to_api_repr()
+        expected = self.FOREIGNTYPEINFO
+        assert result == expected
+
+
 @pytest.mark.parametrize(
     "table_path",
     (