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..8539092
--- /dev/null
+++ b/plugins/pubmed_metadata/dumper.py
@@ -0,0 +1,26 @@
+"""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..137661b
--- /dev/null
+++ b/plugins/pubmed_metadata/mapping.py
@@ -0,0 +1,46 @@
+"""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..63b528a
--- /dev/null
+++ b/plugins/pubmed_metadata/parser.py
@@ -0,0 +1,179 @@
+"""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..0289d91
--- /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 # 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} # nosec B101
+
+
+def test_metadata_fields_are_searchable_and_sortable():
+ properties = mapping_module.get_pubmed_metadata_mapping()["pubmed"]["properties"]
+
+ 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"} # 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 == { # 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
new file mode 100644
index 0000000..cebd203
--- /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 # nosec B101
+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 == { # nosec B101
+ "_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] == [ # nosec B101
+ "PMID:12345678",
+ "PMID:87654321",
+ ]
+ assert documents[0]["pubmed"]["abstract"].endswith("β.") # nosec B101
+
+
+@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 # nosec B101
+
+
+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"] # nosec B101
+
+
+@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..cc77ed7
--- /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 # nosec B101
+STATIC_SPEC.loader.exec_module(static)
+
+
+def test_pinned_release_has_all_expected_shards():
+ 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( # 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
new file mode 100644
index 0000000..ea663c2
--- /dev/null
+++ b/tests/test_pubmed_metadata_uploader.py
@@ -0,0 +1,132 @@
+import gzip
+import importlib
+import json
+import sys
+import types
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def uploader_module(monkeypatch, tmp_path):
+ biothings = types.ModuleType("biothings")
+ 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
+
+ class ParallelizedSourceUploader:
+ pass
+
+ dumper.LastModifiedHTTPDumper = LastModifiedHTTPDumper
+ uploader.ParallelizedSourceUploader = ParallelizedSourceUploader
+ biothings.hub = hub
+ hub.dataload = dataload
+ dataload.dumper = dumper
+ dataload.uploader = uploader
+ 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 # 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
+ )
+
+
+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] == [ # nosec B101
+ "PMID:12345678",
+ "PMID:87654321",
+ ]
+ assert documents[0]["pubmed"]["pub_date"] == "2026-06-30" # nosec B101