Skip to content

cdm_data_loaders.pipelines.ncbi_ftp_promote

NCBI FTP staged file promotion pipeline (Phase 3).

Promotes staged files from the containerized NCBI FTP download pipeline to their final location in the Lakehouse.

PromoteSettings

Bases: LoggerSettings

Configuration for the NCBI FTP file promotion pipeline.

Source code in src/cdm_data_loaders/pipelines/ncbi_ftp_promote.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class PromoteSettings(LoggerSettings):
    """Configuration for the NCBI FTP file promotion pipeline."""

    model_config: SettingsConfigDict = default_settings_with_shortcuts(cli_prog_name=PIPELINE_NAME)

    staging_bucket: PurePosixPath = Field(
        default=DEFAULT_STAGING_BUCKET,
        description="Bucket where staged files are located after download",
    )
    destination_bucket: PurePosixPath = Field(
        default=DEFAULT_DESTINATION_BUCKET,
        description="Bucket to which staged files will be promoted.",
    )
    staging_path: PurePosixPath = Field(
        description="Path to folder in the staging bucket where staged files are located; Should contain `raw_data/` folder",
        validation_alias=AliasChoices("s", "staging_path"),
    )
    destination_path: PurePosixPath = Field(
        default=DEFAULT_DESTINATION_PREFIX,
        description="Path to folder in the destination bucket where files will be promoted to; Will contain `raw_data/` folder",
    )
    removed_manifest_path: Path | None = Field(
        default=None,
        description="Local filesystem path to the removed files manifest from Phase 1, or None to skip archiving removed records",
        validation_alias=AliasChoices("r", "removed-manifest", "removed_manifest"),
    )
    updated_manifest_path: Path | None = Field(
        default=None,
        description="Local filesystem path to the updated files manifest from Phase 1, or None to skip archiving updated records",
        validation_alias=AliasChoices("u", "updated-manifest", "updated_manifest"),
    )
    transfer_manifest_path: PurePosixPath | None = Field(
        default_factory=lambda settings: settings["staging_path"] / DEFAULT_TRANSFER_MANIFEST_FILE,
        description="S3 object key of the transfer manifest to trim after promotion, or None to skip pruning staged files",
        validation_alias=AliasChoices("t", "transfer-manifest", "transfer_manifest"),
    )
    dry_run: CliImplicitFlag[bool] = Field(
        default=False,
        description="Log actions without making changes",
    )

run_promote 

run_promote(config: PromoteSettings) -> None

Main CTS entry point for Phase 3 promotion.

Parameters:

Name Type Description Default
config PromoteSettings

validated promote settings

 required
Source code in src/cdm_data_loaders/pipelines/ncbi_ftp_promote.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def run_promote(config: PromoteSettings) -> None:
    """Main CTS entry point for Phase 3 promotion.

    :param config: validated promote settings
    :type config: PromoteSettings
    """
    report = promote_from_s3(
        staging_bucket=config.staging_bucket,
        staging_key_prefix=config.staging_path,
        lakehouse_bucket=config.destination_bucket,
        lakehouse_key_prefix=config.destination_path,
        removed_manifest_path=config.removed_manifest_path,
        updated_manifest_path=config.updated_manifest_path,
        manifest_s3_key=config.transfer_manifest_path,
        dry_run=config.dry_run,
    )
    if report["failed"] > 0:
        msg = f"Promote completed with {report['failed']} failures"
        raise RuntimeError(msg)