From 1d1ff72959d6ce75de32edececf7682b874d1030 Mon Sep 17 00:00:00 2001 From: shuchenliu Date: Tue, 28 Jul 2026 14:35:54 -0700 Subject: [PATCH 1/2] Add PubMed metadata source --- plugins/pubmed_metadata/README.md | 68 ++++++++++ plugins/pubmed_metadata/__init__.py | 3 + plugins/pubmed_metadata/dumper.py | 27 ++++ plugins/pubmed_metadata/mapping.py | 47 +++++++ plugins/pubmed_metadata/parser.py | 181 +++++++++++++++++++++++++ plugins/pubmed_metadata/static.py | 13 ++ plugins/pubmed_metadata/uploader.py | 56 ++++++++ tests/test_pubmed_metadata_mapping.py | 37 +++++ tests/test_pubmed_metadata_parser.py | 174 ++++++++++++++++++++++++ tests/test_pubmed_metadata_static.py | 20 +++ tests/test_pubmed_metadata_uploader.py | 123 +++++++++++++++++ 11 files changed, 749 insertions(+) create mode 100644 plugins/pubmed_metadata/README.md create mode 100644 plugins/pubmed_metadata/__init__.py create mode 100644 plugins/pubmed_metadata/dumper.py create mode 100644 plugins/pubmed_metadata/mapping.py create mode 100644 plugins/pubmed_metadata/parser.py create mode 100644 plugins/pubmed_metadata/static.py create mode 100644 plugins/pubmed_metadata/uploader.py create mode 100644 tests/test_pubmed_metadata_mapping.py create mode 100644 tests/test_pubmed_metadata_parser.py create mode 100644 tests/test_pubmed_metadata_static.py create mode 100644 tests/test_pubmed_metadata_uploader.py diff --git a/plugins/pubmed_metadata/README.md b/plugins/pubmed_metadata/README.md new file mode 100644 index 0000000..7d3b57c --- /dev/null +++ b/plugins/pubmed_metadata/README.md @@ -0,0 +1,68 @@ +# PubMed metadata + +This source ingests the RENCI `pubmed2db` snapshot published at +. The pinned +release contains 16 gzip-compressed NDJSON shards (about 16.85 GB compressed). + +Each upstream record is stored under a `pubmed` source key in a standalone +PubMed index: + +```json +{ + "_id": "PMID:12345678", + "pubmed": { + "journal": { + "name": "Example Journal", + "abbr": "Example J" + }, + "title": "Example title", + "vol": "1", + "iss": "2", + "pub_date": "2026-06-30", + "abstract": "Example abstract" + } +} +``` + +The parser streams each compressed shard without materializing it in memory. It +requires the exact ten-field upstream schema, string values, valid +`PMID:` identifiers, valid UTF-8, and valid gzip/NDJSON input. A +malformed record fails the upload with the shard and line number rather than +producing a partial or silently altered document. The default storage also +treats duplicate IDs as an error. + +The parser converts NLM month abbreviations to numbers and preserves the +available publication-date precision: `YYYY-MM-DD` when all parts exist, +`YYYY-MM` when the day is missing, and `YYYY` when only the year exists. It +omits `pub_date` when the year is absent. Elasticsearch maps all three forms as +a date; partial dates sort at the beginning of their represented period. + +Downloads and uploads are each capped at four concurrent shards. For +compatibility with pending.api's BioThings Hub 0.12.x dependency, the uploader +partitions the 16 shards among four workers instead of relying only on the +per-source concurrency setting introduced in BioThings Hub 1.x. Abstracts are +retained in Elasticsearch `_source` but are not indexed or sortable. The other +metadata fields are indexed. `title` supports full-text matching and relevance +scoring but is not sortable. `journal.name` uses its `.raw` keyword subfield +when sorting; the keyword and date fields are directly sortable. Because this +is a very large source, the uploader retains only one previous MongoDB source +collection instead of the BioThings default of ten. + +Build `pubmed_metadata` by itself into a versioned `pubmed_*` Elasticsearch +index. After validating the index, point the stable `annotator-pubmed` alias to +it. NodeAnnotator routes `PMID:` identifiers to that alias. + +For subsequent releases, move the alias from the previous index to the newly +validated index in one atomic Elasticsearch alias update. Keep the previous +index temporarily for rollback and remove it separately after validation. +Build configuration and alias state are deployment state and are not stored in +this repository. + +This is a full snapshot rather than an incremental feed. DOI and PMC identifiers +are not present in this export. To adopt a newer snapshot, update `RELEASE` in +`static.py` and verify that its shard count and schema are unchanged. + +The export is produced by +[TranslatorSRI/pubmed2db](https://github.com/TranslatorSRI/pubmed2db) from NLM +PubMed data. Downstream use must follow the +[NLM PubMed terms and conditions](https://ftp.ncbi.nlm.nih.gov/pubmed/baseline/README.txt). diff --git a/plugins/pubmed_metadata/__init__.py b/plugins/pubmed_metadata/__init__.py new file mode 100644 index 0000000..1bad7cf --- /dev/null +++ b/plugins/pubmed_metadata/__init__.py @@ -0,0 +1,3 @@ +# flake8: noqa F401 +from .dumper import PubMedMetadataDumper +from .uploader import PubMedMetadataUploader diff --git a/plugins/pubmed_metadata/dumper.py b/plugins/pubmed_metadata/dumper.py new file mode 100644 index 0000000..fbaefb1 --- /dev/null +++ b/plugins/pubmed_metadata/dumper.py @@ -0,0 +1,27 @@ +"""Dumper for the RENCI PubMed metadata export.""" + +from pathlib import Path + +from biothings import config +from biothings.hub.dataload.dumper import LastModifiedHTTPDumper + +from .static import PUBMED_METADATA_URLS, RELEASE + + +class PubMedMetadataDumper(LastModifiedHTTPDumper): + """Download the pinned PubMed metadata snapshot in bounded parallelism.""" + + SRC_NAME = "pubmed_metadata" + SRC_ROOT_FOLDER = Path(config.DATA_ARCHIVE_ROOT) / SRC_NAME + SRC_URLS = list(PUBMED_METADATA_URLS) + + ARCHIVE = True + AUTO_UPLOAD = True + MAX_PARALLEL_DUMP = 4 + RESOLVE_FILENAME = False + SCHEDULE = None + + def set_release(self) -> None: + """Use the date encoded in the pinned upstream directory.""" + + self.release = RELEASE diff --git a/plugins/pubmed_metadata/mapping.py b/plugins/pubmed_metadata/mapping.py new file mode 100644 index 0000000..8f0a6a7 --- /dev/null +++ b/plugins/pubmed_metadata/mapping.py @@ -0,0 +1,47 @@ +"""Elasticsearch mapping for transformed PubMed metadata documents.""" + +from copy import deepcopy + + +PUBMED_DATE_FORMAT = "strict_date||strict_year_month||strict_year" +MAX_SORTABLE_TEXT_LENGTH = 8191 + + +def _sortable_text() -> dict: + return { + "type": "text", + "fields": { + "raw": { + "type": "keyword", + "ignore_above": MAX_SORTABLE_TEXT_LENGTH, + } + }, + } + + +PUBMED_METADATA_MAPPING = { + "pubmed": { + "properties": { + "journal": { + "properties": { + "name": _sortable_text(), + "abbr": {"type": "keyword"}, + }, + }, + "title": {"type": "text"}, + "vol": {"type": "keyword"}, + "iss": {"type": "keyword"}, + "pub_date": { + "type": "date", + "format": PUBMED_DATE_FORMAT, + }, + "abstract": {"type": "text", "index": False}, + }, + } +} + + +def get_pubmed_metadata_mapping() -> dict: + """Return an independent mapping because Hub consumers may mutate it.""" + + return deepcopy(PUBMED_METADATA_MAPPING) diff --git a/plugins/pubmed_metadata/parser.py b/plugins/pubmed_metadata/parser.py new file mode 100644 index 0000000..f3a90e6 --- /dev/null +++ b/plugins/pubmed_metadata/parser.py @@ -0,0 +1,181 @@ +"""Streaming parser and validation for PubMed metadata NDJSON shards.""" + +import calendar +import gzip +import json +import re +from datetime import date +from pathlib import Path +from typing import Iterator + + +EXPECTED_RECORD_FIELDS = ( + "id", + "journal_name", + "journal_abbrev", + "article_title", + "volume", + "issue", + "pub_year", + "pub_month", + "pub_day", + "abstract", +) +PMID_PATTERN = re.compile(r"^PMID:[1-9][0-9]*$") +YEAR_PATTERN = re.compile(r"^[0-9]{4}$") +NUMERIC_DATE_PART_PATTERN = re.compile(r"^[0-9]{1,2}$") +MONTH_NUMBERS = { + month_abbreviation.lower(): month_number + for month_number, month_abbreviation in enumerate(calendar.month_abbr) + if month_abbreviation +} + + +class PubMedMetadataValidationError(ValueError): + """Raised when an input shard or record does not match the contract.""" + + +def _location(source: str, line_number: int | None) -> str: + if line_number is None: + return source + return f"{source}:{line_number}" + + +def _parse_month(value: str, location: str) -> int: + if NUMERIC_DATE_PART_PATTERN.fullmatch(value): + month = int(value) + else: + month = MONTH_NUMBERS.get(value.lower(), 0) + + if not 1 <= month <= 12: + raise PubMedMetadataValidationError(f"{location}: invalid publication month {value!r}") + return month + + +def _build_pub_date(record: dict, location: str) -> str | None: + year_value = record["pub_year"] + month_value = record["pub_month"] + day_value = record["pub_day"] + + if not year_value: + if month_value or day_value: + raise PubMedMetadataValidationError(f"{location}: publication month/day requires a year") + return None + + if YEAR_PATTERN.fullmatch(year_value) is None or int(year_value) == 0: + raise PubMedMetadataValidationError(f"{location}: invalid publication year {year_value!r}") + + if not month_value: + if day_value: + raise PubMedMetadataValidationError(f"{location}: publication day requires a month") + return year_value + + month = _parse_month(month_value, location) + year_month = f"{year_value}-{month:02d}" + if not day_value: + return year_month + + if NUMERIC_DATE_PART_PATTERN.fullmatch(day_value) is None: + raise PubMedMetadataValidationError(f"{location}: invalid publication day {day_value!r}") + + day = int(day_value) + try: + publication_date = date(int(year_value), month, day) + except ValueError as error: + raise PubMedMetadataValidationError( + f"{location}: invalid publication date {year_value!r}/{month_value!r}/{day_value!r}" + ) from error + return publication_date.isoformat() + + +def transform_pubmed_metadata_record( + record: object, + *, + source: str = "", + line_number: int | None = None, +) -> dict: + """Validate and namespace one upstream PubMed record.""" + + location = _location(source, line_number) + if not isinstance(record, dict): + raise PubMedMetadataValidationError( + f"{location}: expected a JSON object, got {type(record).__name__}" + ) + + expected_fields = set(EXPECTED_RECORD_FIELDS) + actual_fields = set(record) + missing_fields = sorted(expected_fields - actual_fields) + extra_fields = sorted(str(field) for field in actual_fields - expected_fields) + if missing_fields or extra_fields: + details = [] + if missing_fields: + details.append(f"missing fields: {', '.join(missing_fields)}") + if extra_fields: + details.append(f"unexpected fields: {', '.join(extra_fields)}") + raise PubMedMetadataValidationError(f"{location}: {'; '.join(details)}") + + non_string_fields = sorted( + field for field in EXPECTED_RECORD_FIELDS if not isinstance(record[field], str) + ) + if non_string_fields: + raise PubMedMetadataValidationError( + f"{location}: fields must contain strings: {', '.join(non_string_fields)}" + ) + + pubmed_id = record["id"] + if PMID_PATTERN.fullmatch(pubmed_id) is None: + raise PubMedMetadataValidationError(f"{location}: invalid PubMed identifier {pubmed_id!r}") + + pubmed = { + "journal": { + "name": record["journal_name"], + "abbr": record["journal_abbrev"], + }, + "title": record["article_title"], + "vol": record["volume"], + "iss": record["issue"], + "abstract": record["abstract"], + } + pub_date = _build_pub_date(record, location) + if pub_date is not None: + pubmed["pub_date"] = pub_date + + return { + "_id": pubmed_id, + "pubmed": pubmed, + } + + +def iter_pubmed_metadata_documents(data_path: str | Path) -> Iterator[dict]: + """Yield validated documents from a gzip-compressed NDJSON shard.""" + + path = Path(data_path) + try: + with gzip.open( + path, + mode="rt", + encoding="utf-8", + errors="strict", + newline="", + ) as input_file: + for line_number, line in enumerate(input_file, start=1): + if not line.strip(): + raise PubMedMetadataValidationError( + f"{path}:{line_number}: blank lines are not allowed" + ) + try: + record = json.loads(line) + except json.JSONDecodeError as error: + raise PubMedMetadataValidationError( + f"{path}:{line_number}: invalid JSON: {error.msg}" + ) from error + + yield transform_pubmed_metadata_record( + record, + source=str(path), + line_number=line_number, + ) + except PubMedMetadataValidationError: + raise + except (EOFError, OSError, UnicodeError) as error: + raise PubMedMetadataValidationError(f"{path}: unable to read gzip stream: {error}") from error diff --git a/plugins/pubmed_metadata/static.py b/plugins/pubmed_metadata/static.py new file mode 100644 index 0000000..2a03b24 --- /dev/null +++ b/plugins/pubmed_metadata/static.py @@ -0,0 +1,13 @@ +"""Release-specific constants for the PubMed metadata source.""" + +RELEASE = "2026jun30" +BASE_URL = f"https://stars.renci.org/var/babel_outputs/pubmed2db/{RELEASE}/" +SHARD_COUNT = 16 + +PUBMED_METADATA_FILES = tuple( + f"pubmed_metadata_{index:05d}.ndjson.gz" for index in range(SHARD_COUNT) +) +PUBMED_METADATA_URLS = tuple(f"{BASE_URL}{filename}" for filename in PUBMED_METADATA_FILES) + +NLM_TERMS_URL = "https://ftp.ncbi.nlm.nih.gov/pubmed/baseline/README.txt" +PUBMED2DB_URL = "https://github.com/TranslatorSRI/pubmed2db" diff --git a/plugins/pubmed_metadata/uploader.py b/plugins/pubmed_metadata/uploader.py new file mode 100644 index 0000000..ef0ce7a --- /dev/null +++ b/plugins/pubmed_metadata/uploader.py @@ -0,0 +1,56 @@ +"""Uploader for the RENCI PubMed metadata export.""" + +from pathlib import Path + +from biothings.hub.dataload.uploader import ParallelizedSourceUploader + +from .mapping import get_pubmed_metadata_mapping +from .parser import iter_pubmed_metadata_documents +from .static import BASE_URL, NLM_TERMS_URL, PUBMED2DB_URL, PUBMED_METADATA_FILES + + +class PubMedMetadataUploader(ParallelizedSourceUploader): + """Stream each upstream shard into one source collection.""" + + name = "pubmed_metadata" + MAX_PARALLEL_UPLOAD = 4 + keep_archive = 1 + __metadata__ = { + "src_meta": { + "url": BASE_URL, + "license": "NLM PubMed Terms and Conditions", + "license_url": NLM_TERMS_URL, + "description": ( + "PubMed citation metadata exported for Translator by " + f"{PUBMED2DB_URL}" + ), + } + } + + def jobs(self) -> list[tuple[tuple[str, ...]]]: + data_folder = Path(self.data_folder) + shard_paths = [data_folder / filename for filename in PUBMED_METADATA_FILES] + missing_paths = [path.name for path in shard_paths if not path.is_file()] + if missing_paths: + raise FileNotFoundError( + "PubMed metadata upload requires all 16 shards; missing: " + + ", ".join(missing_paths) + ) + + # pending.api still uses BioThings Hub 0.12.x, whose parallel uploader + # does not enforce MAX_PARALLEL_UPLOAD. Partitioning the shards into + # four jobs preserves the source's intended concurrency on both the + # 0.12.x and 1.x Hub implementations. + shard_groups = [ + tuple(str(path) for path in shard_paths[group_index:: self.MAX_PARALLEL_UPLOAD]) + for group_index in range(self.MAX_PARALLEL_UPLOAD) + ] + return [(group,) for group in shard_groups] + + def load_data(self, data_paths: tuple[str, ...]): + for data_path in data_paths: + yield from iter_pubmed_metadata_documents(data_path) + + @classmethod + def get_mapping(cls) -> dict: + return get_pubmed_metadata_mapping() diff --git a/tests/test_pubmed_metadata_mapping.py b/tests/test_pubmed_metadata_mapping.py new file mode 100644 index 0000000..770ec8b --- /dev/null +++ b/tests/test_pubmed_metadata_mapping.py @@ -0,0 +1,37 @@ +import importlib.util +from pathlib import Path + + +MAPPING_PATH = Path(__file__).parents[1] / "plugins" / "pubmed_metadata" / "mapping.py" +MAPPING_SPEC = importlib.util.spec_from_file_location("pubmed_metadata_mapping", MAPPING_PATH) +mapping_module = importlib.util.module_from_spec(MAPPING_SPEC) +assert MAPPING_SPEC.loader is not None +MAPPING_SPEC.loader.exec_module(mapping_module) + + +def test_abstract_is_retrieval_only(): + properties = mapping_module.get_pubmed_metadata_mapping()["pubmed"]["properties"] + + assert properties["abstract"] == {"type": "text", "index": False} + + +def test_metadata_fields_are_searchable_and_sortable(): + properties = mapping_module.get_pubmed_metadata_mapping()["pubmed"]["properties"] + + assert properties["title"] == {"type": "text"} + assert properties["journal"]["properties"]["name"]["fields"]["raw"] == { + "type": "keyword", + "ignore_above": 8191, + } + assert properties["journal"]["properties"]["abbr"] == {"type": "keyword"} + assert properties["vol"] == {"type": "keyword"} + assert properties["iss"] == {"type": "keyword"} + + +def test_publication_date_accepts_available_precision(): + pub_date = mapping_module.get_pubmed_metadata_mapping()["pubmed"]["properties"]["pub_date"] + + assert pub_date == { + "type": "date", + "format": "strict_date||strict_year_month||strict_year", + } diff --git a/tests/test_pubmed_metadata_parser.py b/tests/test_pubmed_metadata_parser.py new file mode 100644 index 0000000..5def107 --- /dev/null +++ b/tests/test_pubmed_metadata_parser.py @@ -0,0 +1,174 @@ +import gzip +import importlib.util +import json +from pathlib import Path + +import pytest + + +PARSER_PATH = Path(__file__).parents[1] / "plugins" / "pubmed_metadata" / "parser.py" +PARSER_SPEC = importlib.util.spec_from_file_location("pubmed_metadata_parser", PARSER_PATH) +parser = importlib.util.module_from_spec(PARSER_SPEC) +assert PARSER_SPEC.loader is not None +PARSER_SPEC.loader.exec_module(parser) + + +def upstream_record(**overrides): + record = { + "id": "PMID:12345678", + "journal_name": "Journal of Examples", + "journal_abbrev": "J Ex", + "article_title": "A useful example", + "volume": "12", + "issue": "3", + "pub_year": "2026", + "pub_month": "6", + "pub_day": "30", + "abstract": "An abstract with Unicode: β.", + } + record.update(overrides) + return record + + +def write_shard(path, lines): + with gzip.open(path, "wt", encoding="utf-8", newline="") as output_file: + output_file.writelines(lines) + + +def test_transform_namespaces_pubmed_metadata(): + document = parser.transform_pubmed_metadata_record(upstream_record()) + + assert document == { + "_id": "PMID:12345678", + "pubmed": { + "journal": { + "name": "Journal of Examples", + "abbr": "J Ex", + }, + "title": "A useful example", + "vol": "12", + "iss": "3", + "abstract": "An abstract with Unicode: β.", + "pub_date": "2026-06-30", + }, + } + + +def test_streams_gzip_ndjson(tmp_path): + shard_path = tmp_path / "pubmed_metadata_00000.ndjson.gz" + records = [upstream_record(), upstream_record(id="PMID:87654321")] + write_shard(shard_path, [json.dumps(record) + "\n" for record in records]) + + documents = list(parser.iter_pubmed_metadata_documents(shard_path)) + + assert [document["_id"] for document in documents] == [ + "PMID:12345678", + "PMID:87654321", + ] + assert documents[0]["pubmed"]["abstract"].endswith("β.") + + +@pytest.mark.parametrize( + ("record", "message"), + [ + ( + {key: value for key, value in upstream_record().items() if key != "abstract"}, + "missing fields: abstract", + ), + ( + upstream_record(unexpected="value"), + "unexpected fields: unexpected", + ), + (upstream_record(id="12345678"), "invalid PubMed identifier"), + (upstream_record(pub_year=2026), "fields must contain strings: pub_year"), + ], +) +def test_rejects_invalid_records(record, message): + with pytest.raises(parser.PubMedMetadataValidationError, match=message): + parser.transform_pubmed_metadata_record(record) + + +@pytest.mark.parametrize( + ("date_parts", "expected_date"), + [ + ({"pub_year": "2026", "pub_month": "Jun", "pub_day": ""}, "2026-06"), + ({"pub_year": "2026", "pub_month": "", "pub_day": ""}, "2026"), + ({"pub_year": "2026", "pub_month": "6", "pub_day": "3"}, "2026-06-03"), + ({"pub_year": "2024", "pub_month": "feb", "pub_day": "29"}, "2024-02-29"), + ], +) +def test_builds_dates_at_available_precision(date_parts, expected_date): + document = parser.transform_pubmed_metadata_record(upstream_record(**date_parts)) + + assert document["pubmed"]["pub_date"] == expected_date + + +def test_omits_date_when_all_components_are_missing(): + document = parser.transform_pubmed_metadata_record( + upstream_record(pub_year="", pub_month="", pub_day="") + ) + + assert "pub_date" not in document["pubmed"] + + +@pytest.mark.parametrize( + ("date_parts", "message"), + [ + ( + {"pub_year": "", "pub_month": "Jun", "pub_day": ""}, + "publication month/day requires a year", + ), + ( + {"pub_year": "2026", "pub_month": "", "pub_day": "15"}, + "publication day requires a month", + ), + ( + {"pub_year": "26", "pub_month": "", "pub_day": ""}, + "invalid publication year", + ), + ( + {"pub_year": "2026", "pub_month": "Smarch", "pub_day": ""}, + "invalid publication month", + ), + ( + {"pub_year": "2026", "pub_month": "Feb", "pub_day": "30"}, + "invalid publication date", + ), + ], +) +def test_rejects_invalid_date_components(date_parts, message): + with pytest.raises(parser.PubMedMetadataValidationError, match=message): + parser.transform_pubmed_metadata_record(upstream_record(**date_parts)) + + +def test_reports_shard_and_line_for_invalid_json(tmp_path): + shard_path = tmp_path / "pubmed_metadata_00000.ndjson.gz" + write_shard(shard_path, [json.dumps(upstream_record()) + "\n", "{broken}\n"]) + + with pytest.raises( + parser.PubMedMetadataValidationError, + match=r"pubmed_metadata_00000\.ndjson\.gz:2: invalid JSON", + ): + list(parser.iter_pubmed_metadata_documents(shard_path)) + + +def test_rejects_blank_lines(tmp_path): + shard_path = tmp_path / "pubmed_metadata_00000.ndjson.gz" + write_shard(shard_path, ["\n"]) + + with pytest.raises( + parser.PubMedMetadataValidationError, + match=r"pubmed_metadata_00000\.ndjson\.gz:1: blank lines", + ): + list(parser.iter_pubmed_metadata_documents(shard_path)) + + +def test_rejects_invalid_gzip(tmp_path): + shard_path = tmp_path / "pubmed_metadata_00000.ndjson.gz" + shard_path.write_bytes(b"not a gzip stream") + + with pytest.raises( + parser.PubMedMetadataValidationError, + match="unable to read gzip stream", + ): + list(parser.iter_pubmed_metadata_documents(shard_path)) diff --git a/tests/test_pubmed_metadata_static.py b/tests/test_pubmed_metadata_static.py new file mode 100644 index 0000000..13d9735 --- /dev/null +++ b/tests/test_pubmed_metadata_static.py @@ -0,0 +1,20 @@ +import importlib.util +from pathlib import Path + + +STATIC_PATH = Path(__file__).parents[1] / "plugins" / "pubmed_metadata" / "static.py" +STATIC_SPEC = importlib.util.spec_from_file_location("pubmed_metadata_static", STATIC_PATH) +static = importlib.util.module_from_spec(STATIC_SPEC) +assert STATIC_SPEC.loader is not None +STATIC_SPEC.loader.exec_module(static) + + +def test_pinned_release_has_all_expected_shards(): + assert static.RELEASE == "2026jun30" + assert static.SHARD_COUNT == 16 + assert static.PUBMED_METADATA_FILES == tuple( + f"pubmed_metadata_{index:05d}.ndjson.gz" for index in range(16) + ) + assert static.PUBMED_METADATA_URLS == tuple( + f"{static.BASE_URL}{filename}" for filename in static.PUBMED_METADATA_FILES + ) diff --git a/tests/test_pubmed_metadata_uploader.py b/tests/test_pubmed_metadata_uploader.py new file mode 100644 index 0000000..1d71aa0 --- /dev/null +++ b/tests/test_pubmed_metadata_uploader.py @@ -0,0 +1,123 @@ +import gzip +import importlib +import json +import sys +import types +from pathlib import Path + +import pytest + + +@pytest.fixture +def uploader_module(monkeypatch): + biothings = types.ModuleType("biothings") + biothings.config = types.SimpleNamespace(DATA_ARCHIVE_ROOT="/tmp/pubmed") + hub = types.ModuleType("biothings.hub") + dataload = types.ModuleType("biothings.hub.dataload") + dumper = types.ModuleType("biothings.hub.dataload.dumper") + uploader = types.ModuleType("biothings.hub.dataload.uploader") + + class LastModifiedHTTPDumper: + pass + + class ParallelizedSourceUploader: + pass + + dumper.LastModifiedHTTPDumper = LastModifiedHTTPDumper + uploader.ParallelizedSourceUploader = ParallelizedSourceUploader + modules = { + "biothings": biothings, + "biothings.hub": hub, + "biothings.hub.dataload": dataload, + "biothings.hub.dataload.dumper": dumper, + "biothings.hub.dataload.uploader": uploader, + } + for name, module in modules.items(): + monkeypatch.setitem(sys.modules, name, module) + + original_plugin_modules = { + name: module + for name, module in tuple(sys.modules.items()) + if name == "plugins.pubmed_metadata" or name.startswith("plugins.pubmed_metadata.") + } + for name in original_plugin_modules: + sys.modules.pop(name) + + yield importlib.import_module("plugins.pubmed_metadata.uploader") + + for name in tuple(sys.modules): + if name == "plugins.pubmed_metadata" or name.startswith("plugins.pubmed_metadata."): + sys.modules.pop(name) + sys.modules.update(original_plugin_modules) + + +def create_shards(data_folder: Path, filenames: tuple[str, ...]) -> None: + for filename in filenames: + data_folder.joinpath(filename).touch() + + +def write_record(path: Path, pubmed_id: str) -> None: + record = { + "id": pubmed_id, + "journal_name": "Journal of Examples", + "journal_abbrev": "J Ex", + "article_title": "A useful example", + "volume": "12", + "issue": "3", + "pub_year": "2026", + "pub_month": "Jun", + "pub_day": "30", + "abstract": "An abstract with Unicode: β.", + } + with gzip.open(path, "wt", encoding="utf-8", newline="") as output_file: + output_file.write(json.dumps(record) + "\n") + + +def test_jobs_partition_shards_across_four_workers(tmp_path, uploader_module): + filenames = uploader_module.PUBMED_METADATA_FILES + create_shards(tmp_path, filenames) + + uploader = uploader_module.PubMedMetadataUploader.__new__( + uploader_module.PubMedMetadataUploader + ) + uploader.data_folder = str(tmp_path) + + jobs = uploader.jobs() + + assert len(jobs) == 4 + assert [len(job[0]) for job in jobs] == [4, 4, 4, 4] + assert jobs[0][0] == tuple(str(tmp_path / filename) for filename in filenames[::4]) + assert sorted(path for job in jobs for path in job[0]) == sorted( + str(tmp_path / filename) for filename in filenames + ) + + +def test_jobs_require_every_shard(tmp_path, uploader_module): + filenames = uploader_module.PUBMED_METADATA_FILES + create_shards(tmp_path, filenames[:-1]) + + uploader = uploader_module.PubMedMetadataUploader.__new__( + uploader_module.PubMedMetadataUploader + ) + uploader.data_folder = str(tmp_path) + + with pytest.raises(FileNotFoundError, match=filenames[-1]): + uploader.jobs() + + +def test_load_data_streams_every_shard_in_a_worker_group(tmp_path, uploader_module): + first_shard = tmp_path / "pubmed_metadata_00000.ndjson.gz" + second_shard = tmp_path / "pubmed_metadata_00004.ndjson.gz" + write_record(first_shard, "PMID:12345678") + write_record(second_shard, "PMID:87654321") + + uploader = uploader_module.PubMedMetadataUploader.__new__( + uploader_module.PubMedMetadataUploader + ) + documents = list(uploader.load_data((str(first_shard), str(second_shard)))) + + assert [document["_id"] for document in documents] == [ + "PMID:12345678", + "PMID:87654321", + ] + assert documents[0]["pubmed"]["pub_date"] == "2026-06-30" From c0337f78c714ddd64c8cd4ae149207879a27687e Mon Sep 17 00:00:00 2001 From: shuchenliu Date: Wed, 29 Jul 2026 16:06:59 -0700 Subject: [PATCH 2/2] Address automated review feedback --- plugins/pubmed_metadata/dumper.py | 1 - plugins/pubmed_metadata/mapping.py | 1 - plugins/pubmed_metadata/parser.py | 2 -- tests/test_pubmed_metadata_mapping.py | 16 ++++++++-------- tests/test_pubmed_metadata_parser.py | 12 ++++++------ tests/test_pubmed_metadata_static.py | 10 +++++----- tests/test_pubmed_metadata_uploader.py | 25 +++++++++++++++++-------- 7 files changed, 36 insertions(+), 31 deletions(-) diff --git a/plugins/pubmed_metadata/dumper.py b/plugins/pubmed_metadata/dumper.py index fbaefb1..8539092 100644 --- a/plugins/pubmed_metadata/dumper.py +++ b/plugins/pubmed_metadata/dumper.py @@ -23,5 +23,4 @@ class PubMedMetadataDumper(LastModifiedHTTPDumper): def set_release(self) -> None: """Use the date encoded in the pinned upstream directory.""" - self.release = RELEASE diff --git a/plugins/pubmed_metadata/mapping.py b/plugins/pubmed_metadata/mapping.py index 8f0a6a7..137661b 100644 --- a/plugins/pubmed_metadata/mapping.py +++ b/plugins/pubmed_metadata/mapping.py @@ -43,5 +43,4 @@ def _sortable_text() -> dict: def get_pubmed_metadata_mapping() -> dict: """Return an independent mapping because Hub consumers may mutate it.""" - return deepcopy(PUBMED_METADATA_MAPPING) diff --git a/plugins/pubmed_metadata/parser.py b/plugins/pubmed_metadata/parser.py index f3a90e6..63b528a 100644 --- a/plugins/pubmed_metadata/parser.py +++ b/plugins/pubmed_metadata/parser.py @@ -95,7 +95,6 @@ def transform_pubmed_metadata_record( line_number: int | None = None, ) -> dict: """Validate and namespace one upstream PubMed record.""" - location = _location(source, line_number) if not isinstance(record, dict): raise PubMedMetadataValidationError( @@ -148,7 +147,6 @@ def transform_pubmed_metadata_record( def iter_pubmed_metadata_documents(data_path: str | Path) -> Iterator[dict]: """Yield validated documents from a gzip-compressed NDJSON shard.""" - path = Path(data_path) try: with gzip.open( diff --git a/tests/test_pubmed_metadata_mapping.py b/tests/test_pubmed_metadata_mapping.py index 770ec8b..0289d91 100644 --- a/tests/test_pubmed_metadata_mapping.py +++ b/tests/test_pubmed_metadata_mapping.py @@ -5,33 +5,33 @@ MAPPING_PATH = Path(__file__).parents[1] / "plugins" / "pubmed_metadata" / "mapping.py" MAPPING_SPEC = importlib.util.spec_from_file_location("pubmed_metadata_mapping", MAPPING_PATH) mapping_module = importlib.util.module_from_spec(MAPPING_SPEC) -assert MAPPING_SPEC.loader is not None +assert MAPPING_SPEC.loader is not None # nosec B101 MAPPING_SPEC.loader.exec_module(mapping_module) def test_abstract_is_retrieval_only(): properties = mapping_module.get_pubmed_metadata_mapping()["pubmed"]["properties"] - assert properties["abstract"] == {"type": "text", "index": False} + assert properties["abstract"] == {"type": "text", "index": False} # nosec B101 def test_metadata_fields_are_searchable_and_sortable(): properties = mapping_module.get_pubmed_metadata_mapping()["pubmed"]["properties"] - assert properties["title"] == {"type": "text"} - assert properties["journal"]["properties"]["name"]["fields"]["raw"] == { + assert properties["title"] == {"type": "text"} # nosec B101 + assert properties["journal"]["properties"]["name"]["fields"]["raw"] == { # nosec B101 "type": "keyword", "ignore_above": 8191, } - assert properties["journal"]["properties"]["abbr"] == {"type": "keyword"} - assert properties["vol"] == {"type": "keyword"} - assert properties["iss"] == {"type": "keyword"} + assert properties["journal"]["properties"]["abbr"] == {"type": "keyword"} # nosec B101 + assert properties["vol"] == {"type": "keyword"} # nosec B101 + assert properties["iss"] == {"type": "keyword"} # nosec B101 def test_publication_date_accepts_available_precision(): pub_date = mapping_module.get_pubmed_metadata_mapping()["pubmed"]["properties"]["pub_date"] - assert pub_date == { + assert pub_date == { # nosec B101 "type": "date", "format": "strict_date||strict_year_month||strict_year", } diff --git a/tests/test_pubmed_metadata_parser.py b/tests/test_pubmed_metadata_parser.py index 5def107..cebd203 100644 --- a/tests/test_pubmed_metadata_parser.py +++ b/tests/test_pubmed_metadata_parser.py @@ -9,7 +9,7 @@ PARSER_PATH = Path(__file__).parents[1] / "plugins" / "pubmed_metadata" / "parser.py" PARSER_SPEC = importlib.util.spec_from_file_location("pubmed_metadata_parser", PARSER_PATH) parser = importlib.util.module_from_spec(PARSER_SPEC) -assert PARSER_SPEC.loader is not None +assert PARSER_SPEC.loader is not None # nosec B101 PARSER_SPEC.loader.exec_module(parser) @@ -38,7 +38,7 @@ def write_shard(path, lines): def test_transform_namespaces_pubmed_metadata(): document = parser.transform_pubmed_metadata_record(upstream_record()) - assert document == { + assert document == { # nosec B101 "_id": "PMID:12345678", "pubmed": { "journal": { @@ -61,11 +61,11 @@ def test_streams_gzip_ndjson(tmp_path): documents = list(parser.iter_pubmed_metadata_documents(shard_path)) - assert [document["_id"] for document in documents] == [ + assert [document["_id"] for document in documents] == [ # nosec B101 "PMID:12345678", "PMID:87654321", ] - assert documents[0]["pubmed"]["abstract"].endswith("β.") + assert documents[0]["pubmed"]["abstract"].endswith("β.") # nosec B101 @pytest.mark.parametrize( @@ -100,7 +100,7 @@ def test_rejects_invalid_records(record, message): def test_builds_dates_at_available_precision(date_parts, expected_date): document = parser.transform_pubmed_metadata_record(upstream_record(**date_parts)) - assert document["pubmed"]["pub_date"] == expected_date + assert document["pubmed"]["pub_date"] == expected_date # nosec B101 def test_omits_date_when_all_components_are_missing(): @@ -108,7 +108,7 @@ def test_omits_date_when_all_components_are_missing(): upstream_record(pub_year="", pub_month="", pub_day="") ) - assert "pub_date" not in document["pubmed"] + assert "pub_date" not in document["pubmed"] # nosec B101 @pytest.mark.parametrize( diff --git a/tests/test_pubmed_metadata_static.py b/tests/test_pubmed_metadata_static.py index 13d9735..cc77ed7 100644 --- a/tests/test_pubmed_metadata_static.py +++ b/tests/test_pubmed_metadata_static.py @@ -5,16 +5,16 @@ STATIC_PATH = Path(__file__).parents[1] / "plugins" / "pubmed_metadata" / "static.py" STATIC_SPEC = importlib.util.spec_from_file_location("pubmed_metadata_static", STATIC_PATH) static = importlib.util.module_from_spec(STATIC_SPEC) -assert STATIC_SPEC.loader is not None +assert STATIC_SPEC.loader is not None # nosec B101 STATIC_SPEC.loader.exec_module(static) def test_pinned_release_has_all_expected_shards(): - assert static.RELEASE == "2026jun30" - assert static.SHARD_COUNT == 16 - assert static.PUBMED_METADATA_FILES == tuple( + assert static.RELEASE == "2026jun30" # nosec B101 + assert static.SHARD_COUNT == 16 # nosec B101 + assert static.PUBMED_METADATA_FILES == tuple( # nosec B101 f"pubmed_metadata_{index:05d}.ndjson.gz" for index in range(16) ) - assert static.PUBMED_METADATA_URLS == tuple( + assert static.PUBMED_METADATA_URLS == tuple( # nosec B101 f"{static.BASE_URL}{filename}" for filename in static.PUBMED_METADATA_FILES ) diff --git a/tests/test_pubmed_metadata_uploader.py b/tests/test_pubmed_metadata_uploader.py index 1d71aa0..ea663c2 100644 --- a/tests/test_pubmed_metadata_uploader.py +++ b/tests/test_pubmed_metadata_uploader.py @@ -9,13 +9,16 @@ @pytest.fixture -def uploader_module(monkeypatch): +def uploader_module(monkeypatch, tmp_path): biothings = types.ModuleType("biothings") - biothings.config = types.SimpleNamespace(DATA_ARCHIVE_ROOT="/tmp/pubmed") + biothings.config = types.SimpleNamespace(DATA_ARCHIVE_ROOT=str(tmp_path / "archive")) hub = types.ModuleType("biothings.hub") dataload = types.ModuleType("biothings.hub.dataload") dumper = types.ModuleType("biothings.hub.dataload.dumper") uploader = types.ModuleType("biothings.hub.dataload.uploader") + biothings.__path__ = [] + hub.__path__ = [] + dataload.__path__ = [] class LastModifiedHTTPDumper: pass @@ -25,6 +28,10 @@ class ParallelizedSourceUploader: dumper.LastModifiedHTTPDumper = LastModifiedHTTPDumper uploader.ParallelizedSourceUploader = ParallelizedSourceUploader + biothings.hub = hub + hub.dataload = dataload + dataload.dumper = dumper + dataload.uploader = uploader modules = { "biothings": biothings, "biothings.hub": hub, @@ -84,10 +91,12 @@ def test_jobs_partition_shards_across_four_workers(tmp_path, uploader_module): jobs = uploader.jobs() - assert len(jobs) == 4 - assert [len(job[0]) for job in jobs] == [4, 4, 4, 4] - assert jobs[0][0] == tuple(str(tmp_path / filename) for filename in filenames[::4]) - assert sorted(path for job in jobs for path in job[0]) == sorted( + assert len(jobs) == 4 # nosec B101 + assert [len(job[0]) for job in jobs] == [4, 4, 4, 4] # nosec B101 + assert jobs[0][0] == tuple( # nosec B101 + str(tmp_path / filename) for filename in filenames[::4] + ) + assert sorted(path for job in jobs for path in job[0]) == sorted( # nosec B101 str(tmp_path / filename) for filename in filenames ) @@ -116,8 +125,8 @@ def test_load_data_streams_every_shard_in_a_worker_group(tmp_path, uploader_modu ) documents = list(uploader.load_data((str(first_shard), str(second_shard)))) - assert [document["_id"] for document in documents] == [ + assert [document["_id"] for document in documents] == [ # nosec B101 "PMID:12345678", "PMID:87654321", ] - assert documents[0]["pubmed"]["pub_date"] == "2026-06-30" + assert documents[0]["pubmed"]["pub_date"] == "2026-06-30" # nosec B101