USE 558 - Write fulltext records to TIMDEX dataset

dfh/cli.py

@@ -1,14 +1,16 @@
 import logging
 import time
-from datetime import timedelta
+from collections.abc import Iterator
+from datetime import datetime, timedelta
 
 import click
 import jsonlines
+from timdex_dataset_api.data_types import DatasetFulltext
 
 from dfh.config import configure_logger, configure_sentry
 from dfh.dspace import warm_dspace_auth
 from dfh.harvest import record_and_fulltext_iter
-from dfh.timdex_dataset import TIMDEXThesesRecords
+from dfh.timdex_dataset import TIMDEXThesesRecords, get_timdex_dataset
 
 logger = logging.getLogger(__name__)
 
@@ -115,26 +117,43 @@ def harvest(
     # warm DSpace API authentication
     warm_dspace_auth()
 
+    # init TIMDEXDataset instance, used for reading and writing
+    timdex_dataset = get_timdex_dataset(dataset_location)
+
     # get iterator of target TIMDEX records + bitstream information
     ttr = TIMDEXThesesRecords(
-        dataset_location=dataset_location,
+        timdex_dataset=timdex_dataset,
         run_id=run_id,
         limit=record_limit,
     )
     records_and_bitstreams = ttr.record_and_bitstream_metadata_iter()
 
-    # get iterator of records + fulltext, ready for writing
+    # get iterator of DatasetFulltext instances, ready for writing
     records_and_fulltexts = record_and_fulltext_iter(
         records_and_bitstreams,
         max_workers=workers,
     )
 
-    if not output_jsonl:
-        raise ValueError(
-            "WIP: Until TIMDEX dataset has new 'fulltexts' "
-            "data type, only JSONL output is supported."
-        )
+    # write to local JSONLines file for debugging
+    if output_jsonl:
+        write_jsonlines_output(output_jsonl, records_and_fulltexts)
+    # default: write to TIMDEX dataset
+    else:
+        timdex_dataset.fulltexts.write(records_and_fulltexts)
+
 
+def write_jsonlines_output(
+    output_jsonl: str,
+    records_and_fulltexts: Iterator[DatasetFulltext],
+) -> None:
+    """Utility function to write to JSONLines for local debugging."""
     with jsonlines.open(output_jsonl, "w") as writer:
         for record in records_and_fulltexts:
-            writer.write(record)
+            output_record = record.to_dict()
+            fulltext = output_record["fulltext"]
+            if isinstance(fulltext, bytes):
+                output_record["fulltext"] = fulltext.decode()
+            timestamp = output_record["fulltext_timestamp"]
+            if isinstance(timestamp, datetime):
+                output_record["fulltext_timestamp"] = timestamp.isoformat()
+            writer.write(output_record)

dfh/config.py

@@ -8,14 +8,16 @@ def configure_logger(logger: logging.Logger, *, verbose: bool) -> str:
     if verbose:
         logging.basicConfig(
             format="%(asctime)s %(levelname)s %(name)s.%(funcName)s() line %(lineno)d: "
-            "%(message)s"
+            "%(message)s",
+            force=True,
         )
         logger.setLevel(logging.DEBUG)
         for handler in logging.root.handlers:
             handler.addFilter(logging.Filter("dfh"))
     else:
         logging.basicConfig(
-            format="%(asctime)s %(levelname)s %(name)s.%(funcName)s(): %(message)s"
+            format="%(asctime)s %(levelname)s %(name)s.%(funcName)s(): %(message)s",
+            force=True,
         )
         logger.setLevel(logging.INFO)
     return (

dfh/harvest.py

@@ -2,10 +2,11 @@
 import threading
 import time
 from collections.abc import Iterator
-from concurrent.futures import FIRST_COMPLETED, Future, ThreadPoolExecutor, wait
 
 import requests
 from dspace_rest_client.client import DSpaceClient
+from joblib import Parallel, delayed
+from timdex_dataset_api.data_types import DatasetFulltext
 
 from dfh.dspace import get_dspace_client, get_presigned_url_for_bitstream
 
@@ -18,52 +19,42 @@
 
 def record_and_fulltext_iter(
     records: Iterator[dict],
-    *,
     max_workers: int = 10,
-    log_progress_interval: int = 10,
-) -> Iterator[dict]:
+    log_progress_interval: int = 1000,
+) -> Iterator[DatasetFulltext]:
     """Yield records with fulltext fetched in parallel.
 
-    Uses ThreadPoolExecutor (main IO bottleneck is network) to generate pre-signed URLs
-    and download bitstream content in parallel.
-
-    The worker function _record_with_fulltext() has built-in retries.  This orchestration
-    function with parallelizes the work is not aware of retries.
+    Uses a threaded worker to generate pre-signed URLs and download bitstream content in
+    parallel.  The worker function _get_record_with_fulltext() has built-in retries.  As
+    such, this orchestration function is not concerned with retries.
     """
-    pending: set[Future[dict]] = set()
-    completed_count = 0
-
-    with ThreadPoolExecutor(max_workers=max_workers) as executor:
-        for record in records:
-            pending.add(executor.submit(_record_with_fulltext, record))
-
-            if len(pending) < max_workers:
-                continue
-
-            done, pending = wait(pending, return_when=FIRST_COMPLETED)
-            for future in done:
-                completed_count += 1
-                yield future.result()
-                if completed_count % log_progress_interval == 0:
-                    logger.debug(f"Extracted fulltext for {completed_count} records.")
-
-        while pending:
-            done, pending = wait(pending, return_when=FIRST_COMPLETED)
-            for future in done:
-                completed_count += 1
-                yield future.result()
-                if completed_count % log_progress_interval == 0:
-                    logger.debug(f"Extracted fulltext for {completed_count} records.")
-
-
-def _record_with_fulltext(
+    parallel_client = Parallel(
+        n_jobs=max_workers,
+        prefer="threads",
+        return_as="generator_unordered",
+    )
+    worker_func = delayed(_get_record_with_fulltext)
+
+    results = parallel_client(worker_func(record) for record in records)
+
+    # log results
+    count = 0
+    for result in results:
+        count += 1
+        if count % log_progress_interval == 0:
+            logger.info(f"Extracted fulltext for {count} records.")
+        yield result
+    logger.info(f"Extraction complete for {count} records.")
+
+
+def _get_record_with_fulltext(
     record: dict,
     *,
     retry_attempts: int = 4,
     initial_backoff_seconds: float = 1.0,
     backoff_factor: float = 2.0,
     timeout_seconds: int = 60,
-) -> dict:
+) -> DatasetFulltext:
     """Return a TIMDEX fulltext record for one source record.
 
     The primary work here is two network requests:
@@ -95,34 +86,36 @@ def _record_with_fulltext(
             # download bitstream content from S3
             response = requests.get(pre_signed_url, timeout=timeout_seconds)
             response.raise_for_status()
-            fulltext = response.text
+            fulltext = response.content
 
             # break out of retries loop if successful
             break
 
         except Exception as exc:
             if attempt == retry_attempts:
                 logger.exception(
-                    f"Max retries of {retry_attempts} encountered, failed to download "
+                    f"Max retries of {retry_attempts} encountered, failed to download. "
+                    f"""timdex_record_id '{record["timdex_record_id"]}', """
                     f"bitstream '{bitstream_uuid}'"
                 )
                 break
 
             sleep_seconds = initial_backoff_seconds * (backoff_factor ** (attempt - 1))
             logger.warning(
-                f"Retrying download for bitstream "
-                f"'{bitstream_uuid}' after attempt {attempt}/{retry_attempts} "
+                f"Retrying download for "
+                f"""timdex_record_id '{record["timdex_record_id"]}', """
+                f"bitstream '{bitstream_uuid}'"
+                f"after attempt {attempt}/{retry_attempts} "
                 f"failed; sleeping {sleep_seconds:.1f} seconds. Cause: {exc}"
             )
             time.sleep(sleep_seconds)
 
-    return {
-        "timdex_record_id": record["timdex_record_id"],
-        "run_id": record["run_id"],
-        "run_record_offset": record["run_record_offset"],
-        "fulltext_bitstream_uuid": bitstream_uuid,
-        "fulltext_bitstream_content": fulltext,
-    }
+    return DatasetFulltext(
+        timdex_record_id=record["timdex_record_id"],
+        run_id=record["run_id"],
+        run_record_offset=record["run_record_offset"],
+        fulltext=fulltext,
+    )
 
 
 def _get_dspace_client_for_thread() -> DSpaceClient:

dfh/timdex_dataset.py

@@ -24,6 +24,20 @@ class TextBitstreamInfo(TypedDict):
     mimetype: str | None
 
 
+def get_timdex_dataset(dataset_location: str | None) -> TIMDEXDataset:
+    """Get an initialized TIMDEXDataset instance.
+
+    If dataset_location is not passed, look to env var TIMDEX_DATASET_LOCATION.
+    """
+    dataset_location = dataset_location or os.getenv("TIMDEX_DATASET_LOCATION")
+    if not dataset_location:
+        raise ValueError(
+            "'dataset_location' must be passed on init "
+            "or env var 'TIMDEX_DATASET_LOCATION' set"
+        )
+    return TIMDEXDataset(dataset_location)
+
+
 class TIMDEXThesesRecords:
     """Class to retrieve TIMDEX dataset records of DSpace Theses.
 
@@ -42,32 +56,21 @@ class TIMDEXThesesRecords:
 
     def __init__(
         self,
-        dataset_location: str | None = None,
+        timdex_dataset: TIMDEXDataset,
         run_id: str | None = None,
         limit: int | None = None,
     ) -> None:
+        self.timdex_dataset = timdex_dataset
         self.run_id = run_id
         self.limit = limit
 
-        self.timdex_dataset = self._init_timdex_dataset(dataset_location)
-
-    def _init_timdex_dataset(self, dataset_location: str | None) -> TIMDEXDataset:
-        """Init TIMDEXDataset and compose to self.
-
-        If dataset_location is not passed, look to env var TIMDEX_DATASET_LOCATION.
-        """
-        dataset_location = dataset_location or os.getenv("TIMDEX_DATASET_LOCATION")
-        if not dataset_location:
-            raise ValueError(
-                "'dataset_location' must be passed on init "
-                "or env var 'TIMDEX_DATASET_LOCATION' set"
-            )
-        return TIMDEXDataset(dataset_location)
-
     def record_and_bitstream_metadata_iter(
         self,
     ) -> Iterator[dict]:
         """Yield TIMDEX DSpace thesis records with their fulltext bitstream info."""
+        logger.debug(
+            "Preparing to yield record + bitstream metadata from TIMDEX dataset."
+        )
         record_count = 0
         error_count = 0
         for record in self.theses_records():

pyproject.toml

@@ -10,6 +10,7 @@ requires-python = ">=3.13"
 dependencies = [
     "click>=8.2.1",
     "dspace-rest-client>=0.1.17",
+    "joblib>=1.5.3",
     "jsonlines>=4.0.0",
     "requests>=2.33.1",
     "sentry-sdk>=2.34.1",
@@ -20,6 +21,7 @@ dependencies = [
 dev = [
     "coveralls>=4.0.1",
     "ipython>=9.13.0",
+    "joblib-stubs>=1.5.3.1.20260117",
     "mypy>=1.17.1",
     "pip-audit>=2.9.0",
     "pre-commit>=4.3.0",
@@ -107,7 +109,7 @@ dfh = "dfh.cli:main"
 include = ["dfh", "dfh.*"]
 
 [tool.uv.sources]
-timdex-dataset-api = { git = "https://github.com/MITLibraries/timdex-dataset-api" }
+timdex-dataset-api = { git = "https://github.com/MITLibraries/timdex-dataset-api", rev = "USE-531-fulltext-data-type-add" }
 
 [build-system]
 requires = ["setuptools>=61"]

tests/conftest.py

@@ -3,6 +3,7 @@
 
 import pytest
 from click.testing import CliRunner
+from timdex_dataset_api import TIMDEXDataset
 
 from dfh.timdex_dataset import TIMDEXThesesRecords
 
@@ -90,5 +91,10 @@ def dspace_mets_text_bitstream_info():
 
 
 @pytest.fixture
-def timdex_theses_records(tmp_path):
-    return TIMDEXThesesRecords(dataset_location=str(tmp_path))
+def timdex_dataset(tmp_path):
+    return TIMDEXDataset(str(tmp_path))
+
+
+@pytest.fixture
+def timdex_theses_records(timdex_dataset):
+    return TIMDEXThesesRecords(timdex_dataset=timdex_dataset)

tests/test_cli.py

@@ -0,0 +1,37 @@
+import json
+
+from timdex_dataset_api.data_types import DatasetFulltext
+
+from dfh.cli import write_jsonlines_output
+
+RUN_RECORD_OFFSET = 12
+
+
+def test_write_jsonlines_output_serializes_dataset_fulltext(tmp_path):
+    output_jsonl = tmp_path / "fulltexts.jsonl"
+    dataset_fulltexts = iter(
+        [
+            DatasetFulltext(
+                timdex_record_id="dspace:123",
+                run_id="run-1",
+                run_record_offset=RUN_RECORD_OFFSET,
+                fulltext_timestamp="2026-05-18T12:34:56+00:00",
+                fulltext=b"fulltext content",
+            ),
+        ]
+    )
+
+    write_jsonlines_output(str(output_jsonl), dataset_fulltexts)
+
+    output_record = json.loads(output_jsonl.read_text().strip())
+    assert output_record == {
+        "timdex_record_id": "dspace:123",
+        "run_id": "run-1",
+        "run_record_offset": RUN_RECORD_OFFSET,
+        "fulltext_timestamp": "2026-05-18T12:34:56+00:00",
+        "fulltext_md5": "1fd0a1296d0665461f688201f63a37f1",
+        "fulltext": "fulltext content",
+        "year": "2026",
+        "month": "05",
+        "day": "18",
+    }

tests/test_dataset.py

@@ -8,15 +8,15 @@
 from dfh.timdex_dataset import (
     MissingTextBitstreamError,
     SourceRecordParseError,
-    TIMDEXThesesRecords,
+    get_timdex_dataset,
 )
 
 
-def test_init_requires_dataset_location(monkeypatch):
+def test_get_timdex_dataset_requires_dataset_location(monkeypatch):
     monkeypatch.delenv("TIMDEX_DATASET_LOCATION", raising=False)
 
     with pytest.raises(ValueError, match="dataset_location"):
-        TIMDEXThesesRecords()
+        get_timdex_dataset(None)
 
 
 def test_theses_records_returns_only_records_with_thesis_content_type(