fedrq.backends.libdnf5.backend

This module contains a fedrq backend (i.e. an implementation of the fedrq.backends.base.BackendMod interface) that uses the libdnf5 Python bindings.

BACKEND module-attribute

BACKEND = 'libdnf5'

RepoError module-attribute

RepoError = RuntimeError

BaseMaker

BaseMaker(
    base: libdnf5.base.Base | None = None,
    *,
    initialized: bool = False,
    config_loaded: bool = False
)

Bases: BaseMakerBase

Create a Base object and load repos

Initialize and configure the base object.

Parameters:

Name	Type	Description	Default
base	Base | None	Pass in a libdnf.base.Base object to configure instead of creating a new one.	None
initialized	bool	Set to True if base.setup() has already been called. Only applies when base is passed.	False
config_loaded	bool	Set to True if base.load_config_from_file() has already been called. Only applies when base is passed.	False

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def __init__(
    self,
    base: libdnf5.base.Base | None = None,
    *,
    initialized: bool = False,
    config_loaded: bool = False,
) -> None:
    """
    Initialize and configure the base object.

    Args:
        base:
            Pass in a [`libdnf.base.Base`][libdnf5.base.Base] object to
            configure instead of creating a new one.
        initialized:
            Set to True if `base.setup()` has already been called. Only
            applies when `base` is passed.
        config_loaded:
            Set to True if base.load_config_from_file() has already been
            called. Only applies when `base` is passed.
    """
    self.base = base or libdnf5.base.Base()
    self.initialized = initialized if base else False
    if not base or not config_loaded:
        # dnf 5.2.0
        try:
            self.base.load_config()
        except AttributeError:
            self.base.load_config_from_file()

backend property

backend: BackendMod

base instance-attribute

base: Base = base or Base()

conf property

conf: ConfigMain

config property

config: ConfigMain

DEPRECATED: use conf property instead

initialized instance-attribute

initialized = initialized if base else False

rs property

rs: RepoSackWeakPtr

vars property

vars: Vars

create_repo

create_repo(repoid: str, **kwargs: t.Any) -> None

Add a Repo object to the repo sack and configure it.

Parameters:

Name	Type	Description	Default
kwargs	Any	key-values options that should be set on the Repo object values (like $basearch) will be substituted automatically.	{}

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def create_repo(self, repoid: str, **kwargs: t.Any) -> None:
    """
    Add a Repo object to the repo sack and configure it.

    Args:
        kwargs:
            key-values options that should be set on the Repo object values
            (like $basearch) will be substituted automatically.
    """
    repo = self.rs.create_repo(repoid)
    config = repo.get_config()
    for key, value in kwargs.items():
        value = self._substitute(value)
        self._set(config, key, value)

disable_repo

disable_repo(
    repo: str, ignore_missing: bool = True
) -> None

Disable a repo by its id. Raise a ValueError if the repoid is not in self.base’s configuration when ignore_missing is False.

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def disable_repo(self, repo: str, ignore_missing: bool = True) -> None:
    """
    Disable a repo by its id.
    Raise a ValueError if the repoid is not in `self.base`'s configuration
    when ignore_missing is False.
    """
    repoq = libdnf5.repo.RepoQuery(self.base)
    repoq.filter_id(repo, libdnf5.common.QueryCmp_GLOB)
    if not ignore_missing and not repoq:
        raise ValueError(f"{repo} repo definition was not found.")
    for result in repoq:
        result.disable()  # type: ignore

enable_repo

enable_repo(repo: str) -> None

Enable a repo by its id. Raise a ValueError if the repoid is not in self.base’s configuration.

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def enable_repo(self, repo: str) -> None:
    """
    Enable a repo by its id.
    Raise a ValueError if the repoid is not in `self.base`'s configuration.
    """
    repoq = libdnf5.repo.RepoQuery(self.base)
    repoq.filter_id(repo, libdnf5.common.QueryCmp_GLOB)
    if not repoq:
        raise ValueError(f"{repo} repo definition was not found.")
    for result in repoq:
        result.enable()  # type: ignore

enable_repos

enable_repos(repos: Collection[str]) -> None

Enable a list of repositories by their repoid. Raise a ValueError if the repoid is not in self.base’s configuration.

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def enable_repos(self, repos: Collection[str]) -> None:
    """
    Enable a list of repositories by their repoid.
    Raise a ValueError if the repoid is not in `self.base`'s configuration.
    """
    for repo in repos:
        self.enable_repo(repo)

enable_source_repos

enable_source_repos() -> None

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def enable_source_repos(self) -> None:
    self.rs.enable_source_repos()

fill_sack

fill_sack(
    *,
    from_cache: bool = False,
    load_system_repo: bool = False
) -> libdnf5.base.Base

Fill the sack and returns the Base object. The repository configuration shouldn’t be manipulated after this.

Note that the _cachedir arg is private and subject to removal.

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def fill_sack(
    self,
    *,
    from_cache: bool = False,
    load_system_repo: bool = False,
) -> libdnf5.base.Base:
    """
    Fill the sack and returns the Base object.
    The repository configuration shouldn't be manipulated after this.

    Note that the `_cachedir` arg is private and subject to removal.
    """
    if from_cache:
        raise NotImplementedError
    try:
        self.rs.load_repos(
            libdnf5.repo.Repo.Type_SYSTEM
            if load_system_repo
            else libdnf5.repo.Repo.Type_AVAILABLE
        )
    except AttributeError:
        self.rs.update_and_load_enabled_repos(load_system_repo)
    return self.base

load_changelogs

load_changelogs(enable: bool = True) -> None

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def load_changelogs(self, enable: bool = True) -> None:
    self._add_metadata_type(libdnf5.conf.METADATA_TYPE_OTHER, enable)

load_filelists

load_filelists(enable: bool = True) -> None

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def load_filelists(self, enable: bool = True) -> None:
    self._add_metadata_type(libdnf5.conf.METADATA_TYPE_FILELISTS, enable)

read_repofile

read_repofile(file: StrPath) -> None

Load repositories from a repo file

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def read_repofile(self, file: StrPath) -> None:
    """
    Load repositories from a repo file
    """
    self.rs.create_repos_from_file(str(file))

read_system_repos

read_system_repos(disable: bool = True) -> None

Load system repositories into the base object. By default, they are all disabled even if ‘enabled=1’ is in the repository configuration.

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def read_system_repos(self, disable: bool = True) -> None:
    """
    Load system repositories into the base object.
    By default, they are all disabled even if 'enabled=1' is in the
    repository configuration.
    """
    self.rs.create_repos_from_system_configuration()
    if disable:
        repoq = libdnf5.repo.RepoQuery(self.base)
        repoq.filter_enabled(True)
        for repo in repoq:
            repo.disable()

repolist

repolist(enabled: bool | None = None) -> list[str]

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def repolist(self, enabled: bool | None = None) -> list[str]:
    repoq = libdnf5.repo.RepoQuery(self.base)
    if enabled is not None:
        repoq.filter_enabled(enabled)
    return [r.get_id() for r in repoq]

set

set(key: str, value: t.Any) -> None

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def set(self, key: str, value: t.Any) -> None:
    # if self.initialized:
    #     raise RuntimeError("The base object has already been initialized")
    LOG.debug("Setting config option %s=%r", key, value)
    self._set(self.conf, key, value)

set_var

set_var(key: str, value: t.Any) -> None

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def set_var(self, key: str, value: t.Any) -> None:
    self.vars.set(key, value)

setup

setup() -> None

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def setup(self) -> None:
    if not self.initialized:
        self.base.setup()
        self.initialized = True

Package

Bases: Package, PackageCompat

DEBUGINFO_SUFFIX class-attribute instance-attribute

DEBUGINFO_SUFFIX = '-debuginfo'

DEBUGSOURCE_SUFFIX class-attribute instance-attribute

DEBUGSOURCE_SUFFIX = '-debugsource'

libdnf5.rpm.Package subclass with strong dnf.package.Package compatability

a property

a: str

arch property

arch: str

buildtime property

buildtime: int

conflicts property

conflicts: Iterable[Reldep5]

debug_name property

debug_name: str

description property

description: str

downloadsize property

downloadsize: int

e property

e: int

enhances property

enhances: Iterable[Reldep5]

epoch property

epoch: int

evr property

evr: str

files property

files: Iterable[str]

from_repo property

from_repo: str

installsize property

installsize: int

installtime property

installtime: int

license property

license: str

location property

location: str

name property

name: str

obsoletes property

obsoletes: Iterable[Reldep5]

packager property

packager: str

provides property

provides: Iterable[Reldep5]

r property

r: str

reason property

reason: str

recommends property

recommends: Iterable[Reldep5]

release property

release: str

repo property

repo: RepoWeakPtr

repoid property

repoid: str

reponame property

reponame: str

requires property

requires: Iterable[Reldep5]

size property

size: int

source_debug_name property

source_debug_name: str

source_name property

source_name: Optional[str]

sourcerpm property

sourcerpm: Optional[str]

suggests property

suggests: Iterable[Reldep5]

summary property

summary: str

supplements property

supplements: Iterable[Reldep5]

url property

url: str

v property

v: str

vendor property

vendor: str

version property

version: str

remote_location

remote_location(
    schemes: Collection[str] | None = (
        "http",
        "ftp",
        "file",
        "https",
    )
) -> str | None

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def remote_location(
    self, schemes: Collection[str] | None = ("http", "ftp", "file", "https")
) -> str | None:
    location = self.location
    if not location:  # pragma: no cover
        return None
    repo_obj = self.repo
    mirrors = repo_obj.get_mirrors() or repo_obj.get_config().baseurl
    if not mirrors:  # pragma: no cover
        return None

    for url in mirrors:
        if not schemes:
            return path_join(url, location)
        scheme = urlparse(url).scheme
        if scheme in schemes:
            return path_join(url, location)
    return None

PackageQuery

Bases: PackageQuery, PackageQueryCompat[Package]

filter

filter(
    **kwargs: Unpack[_QueryFilterKwargs],
) -> PackageQuery

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def filter(  # type: ignore[override]
    self,
    **kwargs: Unpack[_QueryFilterKwargs],
) -> PackageQuery:
    self._filter(**kwargs)
    return self

filter_pkg

filter_pkg(
    pkgs: Iterable[libdnf5.rpm.Package],
    comp: _pkg_comps = libdnf5.common.QueryCmp_EQ,
)

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def filter_pkg(
    self,
    pkgs: Iterable[libdnf5.rpm.Package],
    comp: _pkg_comps = libdnf5.common.QueryCmp_EQ,
    /,
):
    if isinstance(pkgs, (libdnf5.rpm.PackageSet, libdnf5.rpm.PackageQuery)):
        newquery = pkgs
    else:
        if hasattr(self, "__rq__"):  # noqa SIM108
            base = self.__rq__.base
        else:
            base = self.get_base()
        newquery = libdnf5.rpm.PackageSet(base)
        newquery.clear()
        for p in pkgs:
            newquery.add(p)
    if comp == libdnf5.common.QueryCmp_EQ:
        self.intersection(newquery)
    elif comp == libdnf5.common.QueryCmp_NEQ:
        self.difference(newquery)
    else:
        raise ValueError()

filterm

filterm(
    **kwargs: Unpack[_QueryFilterKwargs],
) -> PackageQuery

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def filterm(  # type: ignore[override]
    self,
    **kwargs: Unpack[_QueryFilterKwargs],
) -> PackageQuery:
    self._filter(**kwargs)
    return self

union

union(other: PackageQuery) -> PackageQuery

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def union(self, other: PackageQuery) -> PackageQuery:
    self.update(other)
    return self

Repoquery

Repoquery(base: libdnf5.base.Base)

Bases: RepoqueryBase[Package]

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def __init__(self, base: libdnf5.base.Base) -> None:
    self.base: libdnf5.base.Base = base

backend property

backend: BackendMod

base instance-attribute

base: Base = base

base_arches property

base_arches: set[str]

query

query(
    *, arch: str | Iterable[str] | None = None, **kwargs
) -> PackageQuery

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def query(
    self, *, arch: str | Iterable[str] | None = None, **kwargs
) -> PackageQuery:
    return t.cast(PackageQuery, super().query(arch=arch, **kwargs))

resolve_pkg_specs

resolve_pkg_specs(
    specs: Collection[str],
    resolve: bool = False,
    latest: int | None = None,
    with_src: bool = True,
    *,
    with_filenames: bool | None = None,
    with_provides: bool | None = None,
    resolve_provides: bool | None = None,
    nevra_forms: list[NEVRAForms] | None = None
) -> PackageQuery

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

def resolve_pkg_specs(
    self,
    specs: Collection[str],
    resolve: bool = False,
    latest: int | None = None,
    with_src: bool = True,
    *,
    with_filenames: bool | None = None,
    with_provides: bool | None = None,
    resolve_provides: bool | None = None,
    nevra_forms: list[NEVRAForms] | None = None,
) -> PackageQuery:
    opts = self._get_resolve_options(
        resolve, with_filenames, with_provides, resolve_provides
    )
    settings = libdnf5.base.ResolveSpecSettings()
    _sc(settings, "with_filenames", opts["with_filenames"])
    _sc(settings, "with_provides", opts["with_provides"])
    _sc(settings, "nevra_forms", libdnf5.rpm.VectorNevraForm())
    if nevra_forms:
        v_nevra_forms = libdnf5.rpm.VectorNevraForm()
        for form in nevra_forms:
            v_nevra_forms.append(form)
        _sc(settings, "nevra_forms", v_nevra_forms)
    r_query = self.query(empty=True)
    for spec in specs:
        query = self._query()
        query.resolve_pkg_spec(spec, settings, with_src)
        r_query.union(query)
    if opts["with_provides"]:
        r_query = r_query.union(self.query(provides=specs))
    if opts["with_filenames"]:
        r_query = r_query.union(self.query(file=specs))
    filter_latest(r_query, latest)
    return r_query

get_releasever cached

get_releasever() -> str

Return the system releasever

Source code in src/fedrq/backends/libdnf5/backend/__init__.py

@functools.cache
def get_releasever() -> str:
    """
    Return the system releasever
    """
    # libdnf5 >= 5.0.10
    # https://github.com/rpm-software-management/dnf5/pull/448
    base = libdnf5.base.Base()
    return libdnf5.conf.Vars.detect_release(base.get_weak_ptr(), "/").get()

See also

• libdnf5.base.Base
• libdnf5.rpm.Package
• libdnf5.rpm.PackageQuery