Reduce TIMDEXRunManager and TIMDEXDataset coupling

Why these changes are being introduced:

Formerly, an instance of TIMDEXRunManager expected a TIMDEXDataset on init, where
it would utilize the pyarrow TIMDEXDataset.dataset.  This results in an unneeded
tightly coupling betweent these classes.

How this addresses that need:
* TIMDEXRunManager updated to only expect a pyarrow Dataset

Side effects of this change:
* None

Relevant ticket(s):
* https://mitlibraries.atlassian.net/browse/TIMX-496

Author: ghukill

tests/test_runs.py

@@ -12,19 +12,21 @@
 @pytest.fixture
 def timdex_run_manager(dataset_with_runs_location):
     timdex_dataset = TIMDEXDataset(dataset_with_runs_location)
-    return TIMDEXRunManager(timdex_dataset=timdex_dataset)
+    timdex_dataset.load()
+    return TIMDEXRunManager(dataset=timdex_dataset.dataset)
 
 
 def test_timdex_run_manager_init(dataset_with_runs_location):
     timdex_dataset = TIMDEXDataset(dataset_with_runs_location)
-    timdex_run_manager = TIMDEXRunManager(timdex_dataset=timdex_dataset)
+    timdex_dataset.load()
+    timdex_run_manager = TIMDEXRunManager(dataset=timdex_dataset.dataset)
     assert timdex_run_manager._runs_metadata_cache is None
 
 
 def test_timdex_run_manager_parse_single_parquet_file_success(timdex_run_manager):
     """Parse run metadata from first parquet file in fixture dataset.  We know the details
     of this ETL run in advance given the deterministic fixture that generated it."""
-    parquet_filepath = timdex_run_manager.timdex_dataset.dataset.files[0]
+    parquet_filepath = timdex_run_manager.dataset.files[0]
     run_metadata = timdex_run_manager._parse_run_metadata_from_parquet_file(
         parquet_filepath
     )

timdex_dataset_api/dataset.py

@@ -167,7 +167,7 @@ def load(
         if current_records:
             timdex_run_manager = TIMDEXRunManager(timdex_dataset=self)
 
-            # update paths, limiting by source if set
+            timdex_run_manager = TIMDEXRunManager(dataset=self.dataset)
             self.paths = timdex_run_manager.get_current_parquet_files(
                 source=filters.get("source")
             )

timdex_dataset_api/run.py

@@ -3,25 +3,19 @@
 import concurrent.futures
 import logging
 import time
-from typing import TYPE_CHECKING
 
 import pandas as pd
+import pyarrow.dataset as ds
 import pyarrow.parquet as pq
 
-if TYPE_CHECKING:
-    from timdex_dataset_api.dataset import TIMDEXDataset
-
 logger = logging.getLogger(__name__)
 
 
 class TIMDEXRunManager:
     """Manages and provides access to ETL run metadata from the TIMDEX parquet dataset."""
 
-    def __init__(self, timdex_dataset: "TIMDEXDataset"):
-        self.timdex_dataset: TIMDEXDataset = timdex_dataset
-        if self.timdex_dataset.dataset is None:
-            self.timdex_dataset.load()
-
+    def __init__(self, dataset: ds.Dataset):
+        self.dataset = dataset
         self._runs_metadata_cache: pd.DataFrame | None = None
 
     def clear_cache(self) -> None:
@@ -143,7 +137,7 @@ def _get_parquet_files_run_metadata(self, max_workers: int = 250) -> pd.DataFram
         """
         with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
             futures = []
-            for parquet_filepath in self.timdex_dataset.dataset.files:  # type: ignore[attr-defined]
+            for parquet_filepath in self.dataset.files:  # type: ignore[attr-defined]
                 future = executor.submit(
                     self._parse_run_metadata_from_parquet_file,
                     parquet_filepath,
@@ -181,7 +175,7 @@ def _parse_run_metadata_from_parquet_file(self, parquet_filepath: str) -> dict:
         """
         parquet_file = pq.ParquetFile(
             parquet_filepath,
-            filesystem=self.timdex_dataset.filesystem,
+            filesystem=self.dataset.filesystem,  # type: ignore[attr-defined]
         )
 
         file_meta = parquet_file.metadata.to_dict()