Source code for langchain_google_community.bq_storage_vectorstores.utils
from typing import Any, Dict
from google.cloud.exceptions import NotFound
[docs]def validate_column_in_bq_schema(
columns: dict, column_name: str, expected_types: list, expected_modes: list
) -> None:
"""Validates a column within a BigQuery schema.
Args:
columns: A dictionary of BigQuery SchemaField objects representing
the table schema.
column_name: The name of the column to validate.
expected_types: A list of acceptable data types for the column.
expected_modes: A list of acceptable modes for the column.
Raises:
ValueError: If the column doesn't exist, has an unacceptable type,
or has an unacceptable mode.
"""
if column_name not in columns:
raise ValueError(f"Column {column_name} is missing from the schema.")
column = columns[column_name]
if column.field_type not in expected_types:
raise ValueError(
f"Column {column_name} must be one of the following types: {expected_types}"
)
if column.mode not in expected_modes:
raise ValueError(
f"Column {column_name} must be one of the following modes: {expected_modes}"
)
[docs]def doc_match_filter(document: Dict[str, Any], filter: Dict[str, Any]) -> bool:
for column, value in filter.items():
# ignore fields that are not part of the document
if document.get(column, value) != value:
return False
return True
[docs]def cast_proto_type(column: str, value: Any) -> Any:
if column.startswith("int"):
return int(value)
elif column.startswith("double"):
return float(value)
elif column.startswith("bool"):
return bool(value)
return value
[docs]def check_bq_dataset_exists(client: Any, dataset_id: str) -> bool:
from google.cloud import bigquery # type: ignore[attr-defined]
if not isinstance(client, bigquery.Client):
raise TypeError("client must be an instance of bigquery.Client")
try:
client.get_dataset(dataset_id) # Make an API request.
return True
except NotFound:
return False