Add PDB manifest pipeline

docs/pdb_e2e_walkthrough.md

@@ -0,0 +1,90 @@
+# PDB Pipelines - End-to-End Transfer Walkthrough
+
+## Workflow Overview
+
+Semi-automated PDB transfers involve three distinct pipelines run sequentially:
+* `manifest-generation`: Triggered manually in the Lakehouse.
+    - Downloads current PDB holdings files
+    - Optionally bootstraps current Lakehouse PDB snapshot (if previous snapshot was not saved)
+    - Compares current PDB holdings to Lakehouse PDB snapshot to generate:
+        * `transfer_manifest.txt`: New and updated PDB records to download
+        * `updated_manifest.txt`: Lakehouse records to archive before promoting new downloads
+        * `removed_manifest.txt`: Lakehouse records to archive (will not be replaced with new download)
+        * `missing_dates.txt`: Current PDB holdings for which a last-modified date was not included. (Manual review)
+* `download`: cdm-task-service containerized pipeline.
+    - Downloads records in `transfer_manifest.txt` to staging folder
+    - (possibly verifies file contents against JSONSchema - TBD)
+* `promote`: Triggered manually in the Lakehouse.
+    - Archives PDB records in `updated_manifest.txt` and `removed_manifest.txt`
+    - Promotes staged new downloads to PDB Lakehouse destination
+    - Writes frictionless file descriptors for newly promoted PDB records
+
+### Note for local testing
+
+It is possible to test the end-to-end workflow with a local containerized CEPH test store.
+To start the container:
+```sh
+docker run -d \
+  --name ceph \
+  -p 9000:8080 \
+  -p 9001:8443 \
+  -e RGW_PORT=8080 \
+  -e RGW_ACCESS_KEY=test_access_key \
+  -e RGW_SECRET_KEY=test_access_secret \
+  ghcr.io/kbasetest/ceph-rgw-test-image:0.1.5
+```
+Then, set CEPH credentials as environment variables:
+
+```sh
+export AWS_ENDPOINT_URL=http://localhost:9000
+export AWS_ACCESS_KEY_ID=test_access_key
+export AWS_SECRET_ACCESS_KEY=test_access_secret
+```
+
+Finally, create the standard staging and destination buckets in the test store:
+
+```sh
+uv run python scripts/s3_local.py mb s3://cdm-lake
+uv run python scripts/s3_local.py mb s3://cts
+```
+
+### Package Build
+
+The `cdm-data-loaders` package must be present locally. If you haven't already build it, follow these steps:
+
+```sh
+cd cdm-data-loaders
+uv sync
+source .venv/bin/activate
+uv pip install -e .
+```
+
+## Manifest Generation
+
+Manifest generation is triggered by calling the `pdb_manifest` cli tool. At minimum, you must provide a path
+to the folder the manifests should be written to. If that folder doesn't exist, it will be created:
+
+```sh
+uv run pdb_manifest --output-path out
+```
+
+This will trigger the manifest generation and save the manifest files to `./out/`.
+
+### `pdb_manifest` CLI Tool options
+
+* `--help`: Show full usage (includes some options that are ignored by `pdb_manifest`).
+* `--output-path {path}`: **(Required)** Path to the local folder where manifest files are saved. Created automatically if it does not exist.
+* `--destination-bucket {str}`: Specify a non-standard S3 bucket for the current Lakehouse (or test store) records.
+* `--destination-prefix {str}`: Specify a non-standard S3 key prefix for the current Lakehouse (or test store) records. Must contain `raw_data/`.
+* `--snapshot {str}`: Specify a non-standard path (relative to the `destination-prefix`) for the current snapshot file.
+* `--bootstrap {date}`: Bootstrap a snapshot file using the current Lakehouse (or test store) state, assigning the specified ISO 8601 date as the last-modified date for all records (e.g. `2024-01-01`).
+* `--skip-diff True`: Disable reading of the snapshot file. All current records in the PDB holdings file will be included in the `transfer_manifest.txt`.
+* `--regex {str}` / `-r {str}`: Optionally provide a regex filter expression. Only PDB record ids that pass the filter will be included in manifest files.
+
+## Download
+
+_Coming Soon!_
+
+## Promote
+
+_Coming Soon!_

pyproject.toml

@@ -29,6 +29,7 @@ all_the_bacteria = "cdm_data_loaders.pipelines.all_the_bacteria:cli"
 ncbi_ftp_promote = "cdm_data_loaders.pipelines.ncbi_ftp_promote:cli"
 ncbi_ftp_sync = "cdm_data_loaders.pipelines.ncbi_ftp_download:cli"
 ncbi_rest_api = "cdm_data_loaders.pipelines.ncbi_rest_api:cli"
+pdb_manifest = "cdm_data_loaders.pipelines.pdb_manifest:cli"
 uniprot = "cdm_data_loaders.pipelines.uniprot_kb:cli"
 uniref = "cdm_data_loaders.pipelines.uniref:cli"
 

src/cdm_data_loaders/pdb/constants.py

@@ -0,0 +1,79 @@
+"""Constants, enums, classes, and globals used in PDB piplines."""
+
+import re
+from dataclasses import dataclass
+from enum import StrEnum, auto
+from pathlib import Path, PurePosixPath
+
+DEFAULT_DESTINATION_BUCKET: PurePosixPath = PurePosixPath("cdm-lake")
+DEFAULT_DESTINATION_PREFIX: PurePosixPath = PurePosixPath("tenant-general-warehouse") / "refdata" / "datasets" / "pdb"
+DEFAULT_HOLDINGS_SNAPSHOT: PurePosixPath = PurePosixPath("current_holdings_snapshot.json.gz")
+
+HOLDINGS_BASE_URL = PurePosixPath("files-beta.rcsb.org") / "pub" / "wwpdb" / "pdb" / "holdings"
+
+
+class HoldingsFileTypes(StrEnum):
+    """Types of holdings files used by PDB."""
+
+    CURRENT = auto()
+    LAST_MODIFIED = auto()
+    REMOVED = auto()
+
+
+class HoldingsFileSchemas(StrEnum):
+    """Schema for extracted holdings file data."""
+
+    ID_ONLY = auto()
+    ID_DATE = auto()
+
+
+@dataclass(frozen=True)
+class HoldingsFile:
+    """Filename and schema for a holdings file."""
+
+    filename: Path
+    schema: HoldingsFileSchemas
+
+
+@dataclass(frozen=True)
+class ManifestData:
+    """Data needed to build the manifest files."""
+
+    new: list[str]
+    updated: list[str]
+    removed: list[str]
+    missing_dates: list[str]
+
+
+HOLDINGS_FILES: dict[HoldingsFileTypes, HoldingsFile] = {
+    HoldingsFileTypes.CURRENT: HoldingsFile(
+        filename=Path("current_file_holdings.json.gz"),
+        schema=HoldingsFileSchemas.ID_ONLY,
+    ),
+    HoldingsFileTypes.LAST_MODIFIED: HoldingsFile(
+        filename=Path("released_structures_last_modified_dates.json.gz"),
+        schema=HoldingsFileSchemas.ID_DATE,
+    ),
+    HoldingsFileTypes.REMOVED: HoldingsFile(
+        filename=Path("all_removed_entries.json.gz"),
+        schema=HoldingsFileSchemas.ID_ONLY,
+    ),
+}
+
+# Extended PDB ID: "pdb_" followed by exactly 8 lower-case alphanumeric characters.
+# Classic PDB IDs are [0-9A-Z]{4}; in extended format they are zero-padded to 8
+# chars and lowercased, giving [0-9a-z]{8}
+PDB_ID_RE = re.compile(r"^pdb_[0-9a-z]{8}$", re.IGNORECASE)
+PDB_ID_SEARCH_RE = re.compile(r"pdb_[0-9a-z]{8}", re.IGNORECASE)
+
+
+@dataclass
+class PDBRecord:
+    """A single PDB entry as represented in the holdings inventory.
+
+    :param id: extended PDB ID, e.g., ``"pdb_00001abc"``
+    :param last_modified: ISO-8601 date of last modification, e.g. ``"2024-01-15"``
+    """
+
+    id: str
+    last_modified: str = ""