Skip to content

fedrq.backends.dnf.backend

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

BACKEND module-attribute

BACKEND = 'dnf'

Package module-attribute

Package: type[PackageCompat] = Package

PackageQuery module-attribute

PackageQuery: type[PackageQueryCompat] = Query

RepoError module-attribute

RepoError = RepoError

BaseMaker

BaseMaker(base: dnf.Base | None = None)

Bases: BaseMakerBase

Create a Base object and load repos

Initialize and configure the base object.

Source code in src/fedrq/backends/dnf/backend/__init__.py
def __init__(self, base: dnf.Base | None = None) -> None:
    """
    Initialize and configure the base object.
    """
    self.base: dnf.Base = base or dnf.Base()

backend property

backend: BackendMod

base instance-attribute

base: Base = base or Base()

conf property

conf: MainConf

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/dnf/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.
    """
    self.base.repos.add_new_repo(repoid, self.conf, **kwargs)

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/dnf/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.
    """
    if repo_obj := self.base.repos.get_matching(repo):
        repo_obj.disable()
    elif not ignore_missing:
        raise ValueError(f"{repo} repo definition was not found.")

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/dnf/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.
    """
    if repo_obj := self.base.repos.get_matching(repo):
        repo_obj.enable()
    else:
        raise ValueError(f"{repo} repo definition was not found.")

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/dnf/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/dnf/backend/__init__.py
def enable_source_repos(self) -> None:
    self.base.repos.enable_source_repos()

fill_sack

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

Fill the sack and returns the dnf.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/dnf/backend/__init__.py
def fill_sack(
    self,
    *,
    from_cache: bool = False,
    load_system_repo: bool = False,
) -> dnf.Base:
    """
    Fill the sack and returns the dnf.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:
        self.base.fill_sack_from_repos_in_cache(load_system_repo=load_system_repo)
    else:
        self.base.fill_sack(load_system_repo=load_system_repo)
    return self.base

load_changelogs

load_changelogs(enable: bool = True) -> None
Source code in src/fedrq/backends/dnf/backend/__init__.py
def load_changelogs(self, enable: bool = True) -> None:
    for repo in self.base.repos.iter_enabled():
        repo.load_metadata_other = enable

load_filelists

load_filelists(enable: bool = True) -> None
Source code in src/fedrq/backends/dnf/backend/__init__.py
def load_filelists(self, enable: bool = True) -> None:
    # Old versions of dnf always load filelists
    try:
        types: list[str] = self.conf.optional_metadata_types
    except AttributeError:
        return
    if enable:
        types.append("filelists")
        return
    while "filelists" in types:
        types.remove("filelists")

read_repofile

read_repofile(file: StrPath) -> None
Source code in src/fedrq/backends/dnf/backend/__init__.py
def read_repofile(self, file: StrPath) -> None:
    rr = dnf.conf.read.RepoReader(self.base.conf, None)
    for repo in rr._get_repos(str(file)):
        LOG.debug("Adding %s from %s", repo.id, file)
        self.base.repos.add(repo)

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/dnf/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.base.read_all_repos()
    if not disable:
        return None
    for repo in self.base.repos.iter_enabled():
        repo.disable()

repolist

repolist(enabled: bool | None = None) -> list[str]
Source code in src/fedrq/backends/dnf/backend/__init__.py
def repolist(self, enabled: bool | None = None) -> list[str]:
    if enabled is None:
        return list(self.base.repos)
    return [r.id for r in self.base.repos.values() if r.enabled is bool(enabled)]

set

set(key: str, value: t.Any) -> None
Source code in src/fedrq/backends/dnf/backend/__init__.py
def set(self, key: str, value: t.Any) -> None:
    setattr(self.conf, key, value)

set_var

set_var(key: str, value: t.Any) -> None
Source code in src/fedrq/backends/dnf/backend/__init__.py
def set_var(self, key: str, value: t.Any) -> None:
    if key not in self.base.conf.substitutions:
        raise KeyError(f"{key} is not a valid substitution")
    self.set(key, value)

NEVRAForms

Bases: int, Enum

NA class-attribute instance-attribute

NA = FORM_NA

NAME class-attribute instance-attribute

NAME = FORM_NAME

NEV class-attribute instance-attribute

NEV = FORM_NEV

NEVR class-attribute instance-attribute

NEVR = FORM_NEVR

NEVRA class-attribute instance-attribute

NEVRA = FORM_NEVRA

Repoquery

Repoquery(base: dnf.Base)

Bases: RepoqueryBase[PackageCompat]

Source code in src/fedrq/backends/dnf/backend/__init__.py
def __init__(self, base: dnf.Base) -> None:
    self.base: dnf.Base = base

backend property

backend: BackendMod

base instance-attribute

base: Base = base

base_arches property

base_arches: set[str]

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
)
Source code in src/fedrq/backends/dnf/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,
):
    opts = self._get_resolve_options(
        resolve, with_filenames, with_provides, resolve_provides
    )
    resolve_provides = opts.pop("resolve_provides")
    opts["with_src"] = with_src
    if nevra_forms:
        opts["forms"] = nevra_forms

    query = self.query(empty=True)
    for p in specs:
        subject = dnf.subject.Subject(p).get_best_query(self.base.sack, **opts)
        query = query.union(subject)
    if opts["with_provides"]:
        query = query.union(self.query(provides=specs))
    if opts["with_filenames"]:
        query = query.union(self.query(file=specs))
    filter_latest(query, latest)
    return query

get_changelogs

get_changelogs(package: t.Any) -> Iterator[ChangelogEntry]
Source code in src/fedrq/backends/dnf/backend/__init__.py
def get_changelogs(package: t.Any) -> Iterator[ChangelogEntry]:
    for entry in package.changelogs:
        yield ChangelogEntry(
            text=entry["text"], author=entry["author"], date=entry["timestamp"]
        )

get_releasever cached

get_releasever()

Return the system releasever

Source code in src/fedrq/backends/dnf/backend/__init__.py
@cache
def get_releasever():
    """
    Return the system releasever
    """
    return dnf.rpm.detect_releasever("/")

See also