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

google/cloud/bigquery/_helpers.py

@@ -978,11 +978,11 @@ def _build_resource_from_properties(obj, filter_fields):
  """
  partial = {}
  for filter_field in filter_fields:
- api_field = obj._PROPERTY_TO_API_FIELD.get(filter_field)
+ api_field = _get_sub_prop(obj._PROPERTY_TO_API_FIELD, filter_field)
  if api_field is None and filter_field not in obj._properties:
  raise ValueError("No property %s" % filter_field)
  elif api_field is not None:
- partial[api_field] = obj._properties.get(api_field)
+_set_sub_prop(partial, api_field, _get_sub_prop(obj._properties, api_field))
  else:
  # allows properties that are not defined in the library
  # and properties that have the same name as API resource key

google/cloud/bigquery/schema.py

@@ -15,10 +15,9 @@
 """Schemas for BigQuery tables / queries."""
 
 from __future__ import annotations
-import collections
 import enum
 import typing
-from typing import Any, cast, Dict, Iterable, Optional, Union
+from typing import Any, cast, Dict, Iterable, Optional, Union, Sequence
 
 from google.cloud.bigquery import _helpers
 from google.cloud.bigquery import standard_sql
@@ -489,6 +488,8 @@ def _parse_schema_resource(info):
  Optional[Sequence[google.cloud.bigquery.schema.SchemaField`]:
  A list of parsed fields, or ``None`` if no "fields" key found.
  """
+ if isinstance(info, list):
+ return [SchemaField.from_api_repr(f) for f in info]
  return [SchemaField.from_api_repr(f) for f in info.get("fields", ())]
 
 
@@ -501,40 +502,46 @@ def _build_schema_resource(fields):
  Returns:
  Sequence[Dict]: Mappings describing the schema of the supplied fields.
  """
- return [field.to_api_repr() for field in fields]
+ if isinstance(fields, Sequence):
+ # Input is a Sequence (e.g. a list): Process and return a list of SchemaFields
+ return [field.to_api_repr() for field in fields]
+
+ else:
+ raise TypeError("Schema must be a Sequence (e.g. a list) or None.")
 
 
 def _to_schema_fields(schema):
- """Coerce `schema` to a list of schema field instances.
+ """Coerces schema to a list of SchemaField instances while
+ preserving the original structure as much as possible.
 
  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`.
+ schema (Sequence[Union[ \
+ :class:`~google.cloud.bigquery.schema.SchemaField`, \
+ Mapping[str, Any] \
+ ]
+ ]
+ )::
+ Table schema to convert. Can be a list of SchemaField
+ objects or mappings.
 
  Returns:
-Sequence[:class:`~google.cloud.bigquery.schema.SchemaField`]
+A list of SchemaField objects.
 
  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.
+ TypeError: If schema is not a Sequence.
  """
- for field in schema:
- if not isinstance(field, (SchemaField, collections.abc.Mapping)):
- raise ValueError(
- "Schema items must either be fields or compatible "
- "mapping representations."
- )
 
- return [
- field if isinstance(field, SchemaField) else SchemaField.from_api_repr(field)
- for field in schema
- ]
+ if isinstance(schema, Sequence):
+ # Input is a Sequence (e.g. a list): Process and return a list of SchemaFields
+ return [
+ field
+ if isinstance(field, SchemaField)
+ else SchemaField.from_api_repr(field)
+ for field in schema
+ ]
+
+ else:
+ raise TypeError("Schema must be a Sequence (e.g. a list) or None.")
 
 
 class PolicyTagList(object):

google/cloud/bigquery/table.py

@@ -21,7 +21,8 @@
 import functools
 import operator
 import typing
-from typing import Any, Dict, Iterable, Iterator, List, Optional, Tuple, Union
+from typing import Any, Dict, Iterable, Iterator, List, Optional, Tuple, Union, Sequence
+
 import warnings
 
 try:
@@ -66,6 +67,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
@@ -398,7 +400,7 @@ class Table(_TableBase):
  "partitioning_type": "timePartitioning",
  "range_partitioning": "rangePartitioning",
  "time_partitioning": "timePartitioning",
- "schema": "schema",
+ "schema": ["schema", "fields"],
  "snapshot_definition": "snapshotDefinition",
  "clone_definition": "cloneDefinition",
  "streaming_buffer": "streamingBuffer",
@@ -411,6 +413,7 @@ class Table(_TableBase):
  "max_staleness": "maxStaleness",
  "resource_tags": "resourceTags",
  "external_catalog_table_options": "externalCatalogTableOptions",
+ "foreign_type_info": ["schema", "foreignTypeInfo"],
  }
 
  def __init__(self, table_ref, schema=None) -> None:
@@ -451,8 +454,20 @@ 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"])
+ prop = _helpers._get_sub_prop(
+ self._properties, self._PROPERTY_TO_API_FIELD["schema"]
+ )
  if not prop:
  return []
  else:
@@ -463,10 +478,21 @@ def schema(self, value):
  api_field = self._PROPERTY_TO_API_FIELD["schema"]
 
  if value is None:
- self._properties[api_field] = None
- else:
+ _helpers._set_sub_prop(
+ self._properties,
+ api_field,
+ None,
+ )
+ elif isinstance(value, Sequence):
  value = _to_schema_fields(value)
- self._properties[api_field] = {"fields": _build_schema_resource(value)}
+ value = _build_schema_resource(value)
+ _helpers._set_sub_prop(
+ self._properties,
+ api_field,
+ value,
+ )
+ else:
+ raise TypeError("Schema must be a Sequence (e.g. a list) or None.")
 
  @property
  def labels(self):
@@ -1075,6 +1101,43 @@ 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 = _helpers._get_sub_prop(
+ self._properties, 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):
+ value = value.to_api_repr()
+ _helpers._set_sub_prop(
+ 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/job/test_load.py

@@ -272,7 +272,7 @@ def test_schema_setter_invalid_field(self):
 
  config = LoadJobConfig()
  full_name = SchemaField("full_name", "STRING", mode="REQUIRED")
- with self.assertRaises(ValueError):
+ with self.assertRaises(TypeError):
  config.schema = [full_name, object()]
 
  def test_schema_setter(self):

tests/unit/test_client.py

@@ -2051,7 +2051,7 @@ def test_update_dataset(self):
  ds.labels = LABELS
  ds.access_entries = [AccessEntry("OWNER", "userByEmail", "phred@example.com")]
  ds.resource_tags = RESOURCE_TAGS
-fields = [
+filter_fields = [
  "description",
  "friendly_name",
  "location",
@@ -2065,12 +2065,12 @@ def test_update_dataset(self):
  ) as final_attributes:
  ds2 = client.update_dataset(
  ds,
- fields=fields,
+ fields=filter_fields,
  timeout=7.5,
  )
 
  final_attributes.assert_called_once_with(
- {"path": "/%s" % PATH, "fields": fields}, client, None
+ {"path": "/%s" % PATH, "fields": filter_fields}, client, None
  )
 
  conn.api_request.assert_called_once_with(
@@ -2615,7 +2615,7 @@ def test_update_table_w_schema_None(self):
  self.assertEqual(len(conn.api_request.call_args_list), 2)
  req = conn.api_request.call_args_list[1]
  self.assertEqual(req[1]["method"], "PATCH")
- sent = {"schema": None}
+ sent = {"schema": {"fields": None}}
  self.assertEqual(req[1]["data"], sent)
  self.assertEqual(req[1]["path"], "/%s" % path)
  self.assertEqual(len(updated_table.schema), 0)