feat(community): expose BigQueryLoggerConfig and add dataset check

- Export `BigQueryLoggerConfig` in `langchain_google_community/callbacks/__init__.py` to allow easier configuration.
- Add `_ensure_dataset_exists` check to `BigQueryCallbackHandler` and `AsyncBigQueryCallbackHandler` initialization.
- The handlers now raise a `ValueError` immediately if the target BigQuery dataset does not exist, ensuring fail-fast behavior.

Author: Genesis929

libs/community/langchain_google_community/__init__.py

@@ -18,6 +18,7 @@
 from langchain_google_community.callbacks import (
     AsyncBigQueryCallbackHandler,
     BigQueryCallbackHandler,
+    BigQueryLoggerConfig,
 )
 from langchain_google_community.docai import DocAIParser, DocAIParsingResults
 from langchain_google_community.documentai_warehouse import DocumentAIWarehouseRetriever
@@ -81,6 +82,7 @@
     "AsyncBigQueryCallbackHandler",
     "BigQueryCallbackHandler",
     "BigQueryLoader",
+    "BigQueryLoggerConfig",
     "BigQueryVectorStore",
     "CalendarCreateEvent",
     "CalendarDeleteEvent",

libs/community/langchain_google_community/callbacks/__init__.py

@@ -5,9 +5,11 @@
 from langchain_google_community.callbacks.bigquery_callback import (
     AsyncBigQueryCallbackHandler,
     BigQueryCallbackHandler,
+    BigQueryLoggerConfig,
 )
 
 __all__ = [
     "AsyncBigQueryCallbackHandler",
     "BigQueryCallbackHandler",
+    "BigQueryLoggerConfig",
 ]

libs/community/langchain_google_community/callbacks/bigquery_callback.py

@@ -303,6 +303,20 @@ def _get_bigquery_events_schema(bigquery_module: Any) -> list[Any]:
     ]
 
 
+def _ensure_dataset_exists(
+    client: Any, project_id: str, dataset_id: str, cloud_exceptions: Any
+) -> None:
+    if client is None:
+        raise ValueError("BigQuery client is not initialized.")
+    try:
+        client.get_dataset(dataset_id)
+    except cloud_exceptions.NotFound:
+        raise ValueError(
+            f"Dataset '{dataset_id}' does not exist in project '{project_id}'. "
+            "Please create it before initializing the callback handler."
+        )
+
+
 # ==============================================================================
 # CONFIGURATION
 # ==============================================================================
@@ -1186,7 +1200,10 @@ def __init__(
         self._is_shutting_down: bool = False
         self._setup_lock: asyncio.Lock = asyncio.Lock()
 
-        self.client: Any = None
+        self.client = self.bigquery.Client(project=self.project_id)
+        _ensure_dataset_exists(
+            self.client, self.project_id, self.dataset_id, self.cloud_exceptions
+        )
         self.write_client: Any = None
         self.async_batch_processor: _AsyncBatchProcessor | None = None
         self._executor: ThreadPoolExecutor = ThreadPoolExecutor(max_workers=1)
@@ -1201,10 +1218,6 @@ async def _ensure_started(self) -> None:
                 return
             loop = asyncio.get_running_loop()
 
-            self.client = await loop.run_in_executor(  # type: ignore[func-returns-value]
-                self._executor, lambda: self.bigquery.Client(project=self.project_id)
-            )
-
             full_table_id = (
                 f"{self.project_id}.{self.dataset_id}.{self.config.table_id}"
             )
@@ -1717,7 +1730,10 @@ def __init__(
         self._is_shutting_down: bool = False
         self._setup_lock: threading.Lock = threading.Lock()
 
-        self.client: Any = None
+        self.client = self.bigquery.Client(project=self.project_id)
+        _ensure_dataset_exists(
+            self.client, self.project_id, self.dataset_id, self.cloud_exceptions
+        )
         self.write_client: Any = None
         self.batch_processor: _BatchProcessor | None = None
         self._executor: ThreadPoolExecutor = ThreadPoolExecutor(max_workers=1)
@@ -1731,8 +1747,6 @@ def _ensure_started(self) -> None:
             if self._started:
                 return
 
-            self.client = self.bigquery.Client(project=self.project_id)
-
             full_table_id = (
                 f"{self.project_id}.{self.dataset_id}.{self.config.table_id}"
             )

libs/community/tests/unit_tests/callbacks/test_bigquery_callbacks.py

@@ -671,3 +671,47 @@ def test_sync_log_parsing_error(
         assert (
             f"Failed to parse content: {error_message}" in logged_row["error_message"]
         )
+
+
+def test_sync_init_raises_if_dataset_missing(
+    mock_bigquery_clients: Dict[str, Any],
+) -> None:
+    """Test that sync initialization raises ValueError if the dataset does not exist."""
+    mock_bq_client = mock_bigquery_clients["mock_bq_client"]
+    mock_cloud_exceptions = mock_bigquery_clients["mock_cloud_exceptions"]
+
+    # Simulate dataset not found
+    mock_bq_client.get_dataset.side_effect = mock_cloud_exceptions.NotFound(
+        "Dataset not found"
+    )
+
+    with pytest.raises(ValueError, match="Dataset 'test_dataset' does not exist"):
+        BigQueryCallbackHandler(
+            project_id="test-project",
+            dataset_id="test_dataset",
+            table_id="test_table",
+        )
+
+    mock_bq_client.get_dataset.assert_called_with("test_dataset")
+
+
+def test_async_init_raises_if_dataset_missing(
+    mock_bigquery_clients: Dict[str, Any],
+) -> None:
+    """Test that async initialization raises ValueError if the dataset does not exist."""
+    mock_bq_client = mock_bigquery_clients["mock_bq_client"]
+    mock_cloud_exceptions = mock_bigquery_clients["mock_cloud_exceptions"]
+
+    # Simulate dataset not found
+    mock_bq_client.get_dataset.side_effect = mock_cloud_exceptions.NotFound(
+        "Dataset not found"
+    )
+
+    with pytest.raises(ValueError, match="Dataset 'test_dataset' does not exist"):
+        AsyncBigQueryCallbackHandler(
+            project_id="test-project",
+            dataset_id="test_dataset",
+            table_id="test_table",
+        )
+
+    mock_bq_client.get_dataset.assert_called_with("test_dataset")