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