def _get_source_feature_domains(
    schema_or_domain: Union[schema_pb2.Schema, schema_pb2.StructDomain],
    string_domains: Dict[str, schema_pb2.StringDomain]
) -> Dict[str, common_types.DomainType]:
  """Recursively extracts domains of all source features in the schema."""
  result = {}
  for feature in schema_or_domain.feature:
    domain_info = feature.WhichOneof('domain_info')
    if domain_info == 'struct_domain':
      result.update(
          _get_source_feature_domains(feature.struct_domain, string_domains))
    else:
      domain = _get_domain(feature, string_domains)
      if domain is not None:
        result[feature.name] = domain
  return result
Recursively extracts domains of all source features in the schema.
tensorflow_transform/tf_metadata/schema_utils.py