cdm_data_loaders.ncbi_ftp.promote ¶

def promote_from_s3(  # noqa: PLR0913
    *,
    staging_bucket: PurePosixPath,
    staging_key_prefix: PurePosixPath,
    lakehouse_bucket: PurePosixPath,
    lakehouse_key_prefix: PurePosixPath,
    removed_manifest_path: Path | None = None,
    updated_manifest_path: Path | None = None,
    manifest_s3_key: PurePosixPath | None = None,
    ncbi_release: str | None = None,
    dry_run: bool = False,
) -> dict[str, Any]:
    """Promote files from an S3 staging prefix to the final Lakehouse path.

    Downloads each file to a temp location and re-uploads to the final path
    with MD5 metadata from ``.md5`` sidecar files.

    :param staging_bucket: S3 bucket containing the staged files (e.g. ``"cts"``)
    :type staging_bucket: PurePosixPath
    :param staging_key_prefix: S3 key prefix where CTS output was written
    :type staging_key_prefix: PurePosixPath
    :param lakehouse_bucket: S3 bucket for the final Lakehouse destination (e.g. ``"cdm-lake"``)
    :type lakehouse_bucket: PurePosixPath
    :param lakehouse_key_prefix: S3 key prefix for final Lakehouse locations
    :type lakehouse_key_prefix: PurePosixPath
    :param removed_manifest_path: local path to the removed_manifest file
    :type removed_manifest_path: Path | None
    :param updated_manifest_path: local path to the updated_manifest file
    :type updated_manifest_path: Path | None
    :param manifest_s3_key: S3 object key for transfer_manifest.txt (for trimming)
    :type manifest_s3_key: PurePosixPath | None
    :param ncbi_release: NCBI release version tag for archiving
    :type ncbi_release: str | None
    :param dry_run: if True, log actions without side effects
    :type dry_run: bool
    :return: report dict with counts
    """
    # Get list of objects under the staging prefix
    staged_objects: list[dict[str, Any]] = list_objects(f"{staging_bucket / staging_key_prefix}/")

    # Separate data files from sidecars
    sidecars = {PurePosixPath(k["Key"]) for k in staged_objects if k["Key"].endswith((".crc64nvme", ".md5"))}
    data_files = [PurePosixPath(k["Key"]) for k in staged_objects if PurePosixPath(k["Key"]) not in sidecars]

    logger.info("Found %d data files and %d sidecars in staging", len(data_files), len(sidecars))

    # Archive all affected assemblies BEFORE promoting or deleting
    archived = 0
    for manifest_file, reason, delete in [
        (updated_manifest_path, "updated", False),
        (removed_manifest_path, "replaced_or_suppressed", True),
    ]:
        if manifest_file and Path(str(manifest_file)).is_file():
            archived += _archive_assemblies(
                manifest_file,
                lakehouse_bucket=lakehouse_bucket,
                lakehouse_key_prefix=lakehouse_key_prefix,
                ncbi_release=ncbi_release,
                archive_reason=reason,
                delete_source=delete,
                dry_run=dry_run,
            )

    promoted, failed, descriptors_written, promoted_accessions = _promote_data_files(
        data_files,
        sidecars,
        staging_key_prefix,
        lakehouse_key_prefix,
        staging_bucket,
        lakehouse_bucket,
        dry_run=dry_run,
    )

    # Trim manifest for resumability
    if manifest_s3_key and promoted_accessions and not dry_run:
        _trim_manifest(manifest_s3_key, staging_bucket, promoted_accessions)

    if descriptors_written:
        logger.info("Wrote %d frictionless descriptor(s)", descriptors_written)

    report: dict[str, Any] = {
        "timestamp": datetime.now(UTC).isoformat(),
        "promoted": promoted,
        "archived": archived,
        "failed": failed,
        "dry_run": dry_run,
    }

    logger.info(
        "PROMOTE SUMMARY: %d promoted, %d archived, %d failed%s",
        promoted,
        archived,
        failed,
        " (dry-run)" if dry_run else "",
    )
    return report