tomlquery.py

import functools
import sys

import dnf
import hawkey

DNF_CACHEDIR = "_dnf_cache_dir"
ARCH = "x86_64"
METALINK = "https://mirrors.fedoraproject.org/metalink"
KOJI = "http://kojipkgs.fedoraproject.org"

REPOS = {
    "rawhide": (
        {
            "repoid": "rawhide",
            # 'metalink': f'{METALINK}?repo=rawhide&arch=$basearch',
            "baseurl": [f"{KOJI}/repos/rawhide/latest/$basearch/"],
            "metadata_expire": 60 * 60,
        },
        {
            "repoid": "rawhide-source",
            # 'metalink': f'{METALINK}?repo=rawhide-source&arch=$basearch',
            "baseurl": [f"{KOJI}/repos/rawhide/latest/src/"],
            "metadata_expire": 60 * 60,
        },
    ),
}

# Some deps are only pulled in when those are installed:
DEFAULT_GROUPS = (
    #'buildsys-build',  # for composed repo
    "build",  # for koji repo
)


def name_or_str(thing):
    """
    Useful helper to convert various Hawkey/DNF objects to strings.
    Returns the object's name attribute, falls back to the str representation.
    """
    return getattr(thing, "name", str(thing))


def stringify(lst, separator=", "):
    """
    Converts a list of objects to a single string, using the name_or_str() function.
    If no separator is given, separates the items by comma and space.
    """
    return separator.join(name_or_str(i) for i in lst)


@functools.cache
def _base(repo_key):
    f"""
    Creates a DNF base from repositories defined in REPOS, based on the given key.
    The sack is filled, which can be extremely slow if not already cached on disk in {DNF_CACHEDIR}.
    Cache is never invalidated here, remove the directory manually if needed.
    """
    base = dnf.Base()
    conf = base.conf
    conf.arch = ARCH
    conf.cachedir = DNF_CACHEDIR
    conf.substitutions["releasever"] = "rawhide"
    conf.substitutions["basearch"] = ARCH
    for repo in REPOS[repo_key]:
        base.repos.add_new_repo(conf=conf, skip_if_unavailable=False, **repo)
    base.fill_sack(load_system_repo=False, load_available_repos=True)
    return base


def rawhide_group(group_id):
    """
    Return a rawhide comps group of a given id (a.k.a. name)
    """
    base = _base("rawhide")
    base.read_comps()
    for group in base.comps.groups_by_pattern(group_id):
        if group.id == group_id:
            return group
    raise ValueError(f"No such group {group_id}")


def rawhide_sack():
    """
    A filled sack to perform rawhide repoquries. See base() for details.
    """
    return _base("rawhide").sack


def mandatory_packages_in_group(group_id):
    """
    For given group id (a.k.a. name),
    returns a set of names of mandatory packages in it.
    """
    group = rawhide_group(group_id)
    return {
        p.name for p in group.packages_iter() if p.option_type == dnf.comps.MANDATORY
    }


@functools.lru_cache(maxsize=1)
def mandatory_packages_in_groups(groups=DEFAULT_GROUPS):
    """
    For all group ids,
    returns a single set of names of mandatory packages in any of them.
    """
    all_mandatory_packages = set()
    for group in groups:
        all_mandatory_packages |= mandatory_packages_in_group(group)
    return all_mandatory_packages


def buildrequires_of(package, extra_requires=()):
    """
    Given a hawkey package, returns all buildrequires in their string representations.
    The result is a sorted, deduplicated tuple,
    so it can be hashed as an argument to other cached functions.

    This loads the BuildRequires from the rawhide-source repo,
    note that some packages may have different BuildRequires on different architectures
    and the architecture in the source repo is randomly selected by Koji.
    If you know some package is affected by this,
    you can manually add a hashable collection of extra_requires.

    The package name is searched in the source repo
    and this function only works if exactly 1 package is found.
    If multiple are found, something is wrong with the setup -> RuntimeError.
    If none is found, a package by that name does not exist -> ValueError.
    """
    sack = rawhide_sack()
    return tuple(
        sorted(
            set(str(r) for r in package.requires) | set(str(r) for r in extra_requires)
        )
    )


def resolve_requires(requires, ignore_weak_deps=True):
    """
    Given a hashable collection of requirements,
    resolves all of them and the default buildroot packages in the rawhide repos
    and returns a list of hawkey.Packages (in implicit hawkey order) to be installed.

    If ignore_weak_deps is true (the default), weak dependencies (e.g. Recommends) are ignored,
    which is what happens in mock/Koji as well.

    If hawkey wants to upgrade or erase stuff, something is wrong with the setup -> RuntimeError.
    If hawkey cannot resolve the set, the requires are not installable -> ValueError.
    """
    sack = rawhide_sack()
    goal = hawkey.Goal(sack)
    orig_len = len(requires)
    requires += tuple(mandatory_packages_in_groups())
    for dep in requires:
        selector = hawkey.Selector(sack).set(provides=dep)
        goal.install(select=selector)
    if not goal.run(ignore_weak_deps=ignore_weak_deps):
        raise ValueError(
            f"Cannot resolve {stringify(requires)}: "
            f"{stringify(stringify(p) for p in goal.problem_rules())}"
        )
    if goal.list_upgrades() or goal.list_erasures():
        raise RuntimeError(
            "Got packages to upgrade or erase, that should never happen."
        )
    return goal.list_installs()


def resolve_package(package, ignore_weak_deps=True):
    """
    Given one binary hawkey package, resolve it in the default buildroot
    packages in the rawhide repos and returns a list of hawkey.Packages
    (in implicit hawkey order) to be installed.

    If ignore_weak_deps is true (the default), weak dependencies (e.g. Recommends) are ignored,
    which is what happens in mock/Koji as well.

    If hawkey wants to upgrade or erase stuff, something is wrong with the setup -> RuntimeError.
    If hawkey cannot resolve the set, the requires are not installable -> ValueError.
    """
    sack = rawhide_sack()
    goal = hawkey.Goal(sack)
    for dep in mandatory_packages_in_groups():
        selector = hawkey.Selector(sack).set(provides=dep)
        goal.install(select=selector)
    goal.install(package)
    if not goal.run(ignore_weak_deps=ignore_weak_deps):
        raise ValueError(
            f"Cannot resolve {stringify(requires)}: "
            f"{stringify(stringify(p) for p in goal.problem_rules())}"
        )
    if goal.list_upgrades() or goal.list_erasures():
        raise RuntimeError(
            "Got packages to upgrade or erase, that should never happen."
        )
    return goal.list_installs()


def resolve_buildrequires_of(package, *, extra_requires=(), ignore_weak_deps=True):
    """
    A glue function that takes a package (and optional keyword arguments)
    and returns a resolved list of hawkey.Packages to install.

    See buildrequires_of() and resolve_requires() for details.
    """
    brs = buildrequires_of(package, extra_requires=extra_requires)
    return resolve_requires(brs, ignore_weak_deps=ignore_weak_deps)


def package_by_name(package_name):
    """
    For a given string package name, such as python3-toml
    return the one package object found.
    """
    sack = rawhide_sack()
    pkgs = sack.query().filter(name=package_name, arch__neq="src", latest=1).run()
    if not pkgs:
        raise ValueError(f"No binary RPM called {package_name} found.")
    if len(pkgs) > 1:
        raise RuntimeError(
            f"Too many binary RPMs called {package_name} found: {pkgs!r}"
        )
    return pkgs[0]


def whatrequires(package):
    """
    For a given hawkey package
    return all packages that require it and all packages that buildrequire it
    """
    sack = rawhide_sack()
    pkgs = [package]
    runtime = sack.query().filter(requires=pkgs, arch__neq="src").run()
    buildtime = sack.query().filter(requires=pkgs, arch="src").run()
    return runtime, buildtime


if __name__ == "__main__":
    package = package_by_name("python3-toml")
    requires, brs = whatrequires(package)
    print("Run-time dependents:")
    for dependent in requires:
        try:
            if package in resolve_package(dependent):
                print(name_or_str(dependent))
        except ValueError:
            print(name_or_str(dependent), "???")
    print()
    print("Build-time dependents:")
    for dependent in brs:
        try:
            if package in resolve_buildrequires_of(dependent):
                print(name_or_str(dependent))
        except ValueError:
            print(name_or_str(dependent), "???")