Small script to download RPMs from the Chef Community Download API

chef-download.py

#!/usr/bin/python3

import argparse
from typing import Dict, List

import requests


def download_file(url: str, local_filename: str, params: Dict[str, str] = None) -> str:
    """
    https://stackoverflow.com/a/16696317
    """
    # NOTE the stream=True parameter below
    with requests.get(url, stream=True, params=params) as r:
        r.raise_for_status()
        with open(local_filename, "wb") as f:
            for chunk in r.iter_content(chunk_size=8192):
                # If you have chunk encoded response uncomment if
                # and set chunk_size parameter to None.
                # if chunk:
                f.write(chunk)
    return local_filename


def architectures() -> List[str]:
    """
    Returns all architectures that chef offers potentially.
    """
    r = requests.get("https://chefdownload-community.chef.io/architectures")
    return r.json()


def platforms() -> Dict[str, str]:
    """
    Returns all platform IDs with their names.
    """
    r = requests.get("https://chefdownload-community.chef.io/platforms")
    return r.json()


def products(eol: bool = False) -> List[str]:
    """
    Returns all products.

    :param eol: If this is set to true, returns also end of life products.
    """
    r = requests.get(
        "https://chefdownload-community.chef.io/products", params={"eol": eol}
    )
    return r.json()


def packages(product: str, product_version: str, license_id: str):
    """ """
    r = requests.get(
        f"https://chefdownload-community.chef.io/stable/{product}/packages",
        params={"v": product_version, "license_id": license_id},
    )
    return r.json()


def versions_all(product: str, license_id: str) -> List[str]:
    """ """
    r = requests.get(
        f"https://chefdownload-community.chef.io/stable/{product}/versions/all",
        params={"license_id": license_id},
    )
    return r.json()


def versions_latest(product: str, license_id: str) -> str:
    """ """
    r = requests.get(
        f"https://chefdownload-community.chef.io/stable/{product}/versions/latest",
        params={"license_id": license_id},
    )
    return r.json()


def metadata(
    product: str,
    platform_name: str,
    platform_version: str,
    architecture: str,
    product_version: str,
    license_id: str,
) -> Dict[str, str]:
    """ """
    r = requests.get(
        f"https://chefdownload-community.chef.io/stable/{product}/metadata",
        params={
            "p": platform_name,
            "pv": platform_version,
            "m": architecture,
            "v": product_version,
            "license_id": license_id,
        },
    )
    return r.json()


def download_package(
    product: str,
    platform_name: str,
    platform_version: str,
    architecture: str,
    product_version: str,
    license_id: str,
    file_name: str = "chef.rpm",
) -> None:
    """ """
    download_file(
        f"https://chefdownload-community.chef.io/stable/{product}/download",
        file_name,
        {
            "p": platform_name,
            "pv": platform_version,
            "m": architecture,
            "v": product_version,
            "license_id": license_id,
        },
    )


def fileName(
    product: str,
    platform_name: str,
    platform_version: str,
    architecture: str,
    product_version: str,
    license_id: str,
) -> str:
    """ """
    r = requests.get(
        f"https://chefdownload-community.chef.io/stable/{product}/fileName",
        params={
            "p": platform_name,
            "pv": platform_version,
            "m": architecture,
            "v": product_version,
            "license_id": license_id,
        },
    )
    return r.json().get("fileName")


def download_nice(
    product: str,
    platform_name: str,
    platform_version: str,
    architecture: str,
    product_version: str,
    license_id: str,
) -> None:
    """ """
    archs = architectures()
    if architecture not in archs:
        print("Architecture couldn't be identified.")
        return
    platform = platforms()
    if platform_name not in platform.keys():
        print("Platform couldn't be identified.")
        return
    prods = products()
    if product not in prods:
        print("Product couldn't be identified.")
        return
    if product_version is None:
        product_version = versions_latest(product=product, license_id=license_id)
        print(f"Set product version to latest one: {product_version}.")
    else:
        vers = versions_all(product=product, license_id=license_id)
        if product_version not in vers:
            print("Version couldn't be identified.")
            return
    rpm_name = fileName(
        product=product,
        platform_name=platform_name,
        platform_version=platform_version,
        architecture=architecture,
        product_version=product_version,
        license_id=license_id,
    )
    download_package(
        product=product,
        platform_name=platform_name,
        platform_version=platform_version,
        architecture=architecture,
        product_version=product_version,
        license_id=license_id,
        file_name=rpm_name,
    )


def main():
    """
    Main entrypoint for the chef community api downloader.
    """
    parser = argparse.ArgumentParser(
        description="Download Script to obtain information from the Chef Community API."
    )
    parser.add_argument(
        "--endpoint",
        required=True,
        choices=[
            "architectures",
            "platforms",
            "products",
            "packages",
            "version",
            "versions",
            "metadata",
            "download",
            "fileName",
            "download-nice",
        ],
    )
    parser.add_argument("--eol", action="store_true")
    parser.add_argument("--product")
    parser.add_argument("--version")
    parser.add_argument("--license-id")
    parser.add_argument("--platform-name")
    parser.add_argument("--platform-version")
    parser.add_argument("--architecture")
    parser.add_argument("--product-version")
    args = parser.parse_args()
    if args.endpoint == "architectures":
        print(architectures())
    elif args.endpoint == "platforms":
        print(platforms())
    elif args.endpoint == "products":
        print(products(eol=args.eol))
    elif args.endpoint == "packages":
        print(
            packages(
                product=args.product,
                product_version=args.product_version,
                license_id=args.license_id,
            )
        )
    elif args.endpoint == "version":
        versions_latest(product=args.product, license_id=args.license_id)
    elif args.endpoint == "versions":
        versions_all(product=args.product, license_id=args.license_id)
    elif args.endpoint == "metadata":
        print(
            metadata(
                product=args.product,
                platform_name=args.platform_name,
                platform_version=args.platform_version,
                architecture=args.architecture,
                product_version=args.product_version,
                license_id=args.license_id,
            )
        )
    elif args.endpoint == "download":
        download_package(
            product=args.product,
            platform_name=args.platform_name,
            platform_version=args.platform_version,
            architecture=args.architecture,
            product_version=args.product_version,
            license_id=args.license_id,
        )
    elif args.endpoint == "fileName":
        print(
            fileName(
                product=args.product,
                platform_name=args.platform_name,
                platform_version=args.platform_version,
                architecture=args.architecture,
                product_version=args.product_version,
                license_id=args.license_id,
            )
        )
    elif args.endpoint == "download-nice":
        download_nice(
            product=args.product,
            platform_name=args.platform_name,
            platform_version=args.platform_version,
            architecture=args.architecture,
            product_version=args.product_version,
            license_id=args.license_id,
        )
    else:
        print("Nothing implemented yet!")


if __name__ == "__main__":
    main()