Extract dataset checker function into common utils

Author: Bill-hbrhbr

components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py

@@ -163,6 +163,28 @@ def fetch_existing_datasets(
     return {row["name"] for row in rows}
 
 
+def validate_dataset(
+    db_cursor,
+    table_prefix: str,
+    dataset: str,
+    existing_datasets: Set[str] | None = None,
+) -> bool:
+    """
+    Checks if a dataset currently exists in the metadata or in the local dataset cache.
+
+    :param db_cursor:
+    :param table_prefix:
+    :param dataset: The dataset to validate.
+    :param existing_datasets: Returns a refreshed cache of dataset names fetched from the metadata
+                              if the current cache doesn not contain the provided dataset and a
+                              lookup is required.
+    """
+    if existing_datasets is not None and dataset in existing_datasets:
+        return True
+    existing_datasets = fetch_existing_datasets(db_cursor, table_prefix)
+    return dataset in existing_datasets
+
+
 def create_metadata_db_tables(db_cursor, table_prefix: str, dataset: str | None = None) -> None:
     """
     Creates the standard set of tables for CLP's metadata.

components/job-orchestration/job_orchestration/scheduler/query/query_scheduler.py

@@ -41,7 +41,7 @@
     TAGS_TABLE_SUFFIX,
 )
 from clp_py_utils.clp_logging import get_logger, get_logging_formatter, set_logging_level
-from clp_py_utils.clp_metadata_db_utils import fetch_existing_datasets
+from clp_py_utils.clp_metadata_db_utils import validate_dataset
 from clp_py_utils.core import read_yaml_config_file
 from clp_py_utils.decorators import exception_default_value
 from clp_py_utils.sql_adapter import SQL_Adapter
@@ -614,18 +614,6 @@ def dispatch_job_and_update_db(
     )
 
 
-def _validate_dataset(
-    db_cursor,
-    table_prefix: str,
-    dataset: str,
-    existing_datasets: Set[str],
-) -> bool:
-    if dataset in existing_datasets:
-        return True
-    existing_datasets = fetch_existing_datasets(db_cursor, table_prefix)
-    return dataset in existing_datasets
-
-
 def handle_pending_query_jobs(
     db_conn_pool,
     clp_metadata_db_conn_params: Dict[str, any],
@@ -656,7 +644,7 @@ def handle_pending_query_jobs(
             table_prefix = clp_metadata_db_conn_params["table_prefix"]
             if StorageEngine.CLP_S == clp_storage_engine:
                 dataset = QueryJobConfig.parse_obj(job_config).dataset
-                if not _validate_dataset(db_cursor, table_prefix, dataset, existing_datasets):
+                if not validate_dataset(db_cursor, table_prefix, dataset, existing_datasets):
                     logger.error(f"Dataset `{dataset}` doesn't exist.")
                     if not set_job_or_task_status(
                         db_conn,