def snapshot_repo()

in antlir/rpm/snapshot_repo.py [0:0]

• 66 lines of code
• 1 McCabe index (conditional complexity)

def snapshot_repo(argv) -> None:
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    add_standard_args(parser)
    parser.add_argument(
        "--repo-universe",
        required=True,
        help="This is explained in the `repo_db.py` docblock. In production, "
        "it is important for the universe name to match existing "
        "conventions -- DO NOT JUST MAKE ONE UP.",
    )
    parser.add_argument(
        "--repo-name",
        required=True,
        help="Used to distinguish this repo's metadata from others' in the DB.",
    )
    parser.add_argument(
        "--repo-url",
        required=True,
        help="The base URL of the repo -- the part before repodata/repomd.xml. "
        "Supported protocols include file://, https://, and http://.",
    )
    parser.add_argument(
        "--gpg-url",
        required=True,
        action="append",
        help="(May be repeated) Yum will need to import this key to gpgcheck "
        "the repo. To avoid placing blind trust in these keys (e.g. in "
        "case this is an HTTP URL), they are verified against "
        "`--gpg-key-allowlist-dir`",
    )
    args = Path.parse_args(parser, argv)

    init_logging(debug=args.debug)

    # pyre-fixme[16]: `Path` has no attribute `__enter__`.
    with populate_temp_dir_and_rename(
        args.snapshot_dir,
        overwrite=True
        # pyre-fixme[16]: `Iterable` has no attribute `__enter__`.
    ) as td, RepoSnapshot.add_sqlite_to_storage(
        # pyre-fixme[6]: For 1st param expected `Storage` but got `Pluggable`.
        Storage.from_json(args.storage),
        td,
    ) as sqlite_db:
        sizer = RepoSizer()
        snapshot_gpg_keys(
            key_urls=args.gpg_url,
            allowlist_dir=args.gpg_key_allowlist_dir,
            snapshot_dir=td,
        )
        repo = YumDnfConfRepo(
            name=args.repo_name,
            base_url=args.repo_url,
            gpg_key_urls=args.gpg_url,
        )
        _, snapshot = next(
            download_repos(
                repos_and_universes=[(repo, args.repo_universe)],
                cfg=DownloadConfig(
                    db_cfg=args.db,
                    storage_cfg=args.storage,
                    rpm_shard=args.rpm_shard,
                    threads=args.threads,
                ),
            )
        )
        snapshot.visit(sizer).to_sqlite(args.repo_name, sqlite_db)
        log.info(sizer.get_report(f"This {args.rpm_shard} snapshot weighs"))