Depsolving multiple transactions in a row using DNF (version 2)

dnf-depsolving.py

#!/usr/bin/python3

import os
import tempfile

import dnf
import pprint


DEPSOLVE_REQ = {
    "transactions": [
        {
            "package-specs": [
                "@Server",
                "NetworkManager",
                "kernel",
                "kernel-core",
                "kernel-modules",
                "selinux-policy-targeted",
                "efibootmgr",
                "lvm2",
                "grub2-efi-x64",
                "shim-x64",
                "dracut-config-generic",
                "dracut-norescue",
                "bzip2",
                "langpacks-en",
                "grub2-pc",
                "rhc",
                "yum-utils",
                "WALinuxAgent",
                "cloud-init",
                "cloud-utils-growpart",
                "gdisk",
                "hyperv-daemons",
                "nvme-cli",
                "cryptsetup-reencrypt",
                "uuid",
                "rng-tools",
                "patch",
                "dracut-config-generic",
                "grub2-pc",
                "dracut-config-generic",
                "efibootmgr",
                "grub2-efi-x64",
                "shim-x64",
                "kernel"
            ],
            "exclude-specs": [
                "aic94xx-firmware",
                "alsa-firmware",
                "alsa-lib",
                "alsa-sof-firmware",
                "alsa-tools-firmware",
                "dracut-config-rescue",
                "ivtv-firmware",
                "iwl1000-firmware",
                "iwl100-firmware",
                "iwl105-firmware",
                "iwl135-firmware",
                "iwl2000-firmware",
                "iwl2030-firmware",
                "iwl3160-firmware",
                "iwl3945-firmware",
                "iwl4965-firmware",
                "iwl5000-firmware",
                "iwl5150-firmware",
                "iwl6000-firmware",
                "iwl6000g2a-firmware",
                "iwl6000g2b-firmware",
                "iwl6050-firmware",
                "iwl7260-firmware",
                "libertas-sd8686-firmware",
                "libertas-sd8787-firmware",
                "libertas-usb8388-firmware",
                "glibc-all-langpacks",
                "biosdevname",
                "cockpit-podman",
                "bolt",
                "buildah",
                "containernetworking-plugins",
                "dnf-plugin-spacewalk",
                "iprutils",
                "plymouth",
                "podman",
                "python3-dnf-plugin-spacewalk",
                "python3-rhnlib",
                "python3-hwdata",
                "NetworkManager-config-server",
                "rhn-client-tools",
                "rhn-setup",
                "rhnsd",
                "rhn-check",
                "rhnlib",
                "usb_modeswitch"
            ],
            "repos": [
                {
                    "id": "baseos",
                    "baseurl": "https://rpmrepo.osbuild.org/v2/mirror/rhvpn/el8/el8-x86_64-baseos-n8.6-20220301/",
                },
                {
                    "id": "appstream",
                    "baseurl": "https://rpmrepo.osbuild.org/v2/mirror/rhvpn/el8/el8-x86_64-appstream-n8.6-20220301/",
                }
            ]
        },
        {
            "package-specs": [
                "kernel",
                "bind"
            ],
            "exclude-specs": [],
            "repos": [
                {
                    "id": "baseos",
                    "baseurl": "https://rpmrepo.osbuild.org/v2/mirror/rhvpn/el8/el8-x86_64-baseos-n8.6-20220301/",
                },
                {
                    "id": "appstream",
                    "baseurl": "https://rpmrepo.osbuild.org/v2/mirror/rhvpn/el8/el8-x86_64-appstream-n8.6-20220301/",
                }
                # user-defined repos go here...
            ]
        }
    ],
    "module_platform_id": "platform:el8",
    "arch": "x86_64"
}


def CreateDNFBase(repos, module_platform_id, persistdir, cachedir, arch):
    base = dnf.Base()

    base.conf.fastestmirror = True
    base.conf.zchunk = False
    base.conf.module_platform_id = module_platform_id
    base.conf.config_file_path = "/dev/null"
    base.conf.persistdir = persistdir
    base.conf.cachedir = cachedir
    base.conf.substitutions['arch'] = arch
    base.conf.substitutions['basearch'] = dnf.rpm.basearch(arch)

    for repo in repos:
        dnf_repo = dnf.repo.Repo(repo["id"], base.conf)
        dnf_repo.baseurl = repo["baseurl"]
        base.repos.add(dnf_repo)

    base.fill_sack(load_system_repo=False)

    return base


def DepsolveTransaction(dnf_base, installed_pkgs, include_specs, exclude_specs):
    # mark installed packages for installation
    # the resulting transaction will contain all of these packages plus those
    # depsolved based on the provided "install_specs"
    for installed_pkg in installed_pkgs:
        dnf_base.package_install(installed_pkg, strict=True)

    dnf_base.install_specs(include_specs, exclude_specs)
    dnf_base.resolve()

    result = []

    for tsi in dnf_base.transaction:
        if tsi.action not in dnf.transaction.FORWARD_ACTIONS:
            continue
        result.append(tsi.pkg)
    
    return result


def main():
    transactions = []

    for transaction in DEPSOLVE_REQ["transactions"]:
        with tempfile.TemporaryDirectory() as tempdir:
            persistent_dir = os.path.join(tempdir, "persistent")
            os.makedirs(persistent_dir)
            cache_dir = os.path.join(tempdir, "cache")
            os.makedirs(cache_dir)

            # take the package list from the last transaction and treat it as "installed"
            # by directly marking the packages for installation
            installed_pkgs = transactions[-1] if len(transactions) else []

            dnf_base = CreateDNFBase(transaction["repos"], DEPSOLVE_REQ["module_platform_id"], persistent_dir, cache_dir, DEPSOLVE_REQ["arch"])
            transaction_result = DepsolveTransaction(dnf_base, installed_pkgs, transaction["package-specs"], transaction["exclude-specs"])
            transactions.append(transaction_result)

    for idx, transaction in enumerate(transactions):
        print(f"Transaction #{idx}:\nNumber of pkgs in transaction: {len(transaction)}")
        pprint.pprint([f"{p.name}-{p.version}.{p.release}" for p in transaction])
        
        if idx > 0:
            print()
            print(f"Added packages on top of the previous transactions:")
            diff = set(transaction)-set.union(*[set(t) for t in transactions[:idx]])
            pprint.pprint([f"{p.name}-{p.version}.{p.release}" for p in diff])

        print()


if __name__ == "__main__":
    main()