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.

Modules:

  • experimental –

    This module (and the other experimental modules in fedrq) are not meant for public use.

Classes:

Functions:

Attributes:

BACKEND module-attribute 

BACKEND = 'dnf'

Name of the backend

Package module-attribute 

Package: type[PackageCompat] = Package

Annotated alias to dnf.package.Package

PackageQuery module-attribute 

PackageQuery: type[PackageQueryCompat] = Query

Annotated alias to dnf.query.Query

PackageQueryAlias module-attribute 

PackageQueryAlias: TypeAlias = PackageQueryCompat[
    PackageCompat
]

RepoError module-attribute 

RepoError = RepoError

Error when loading repositories

BaseMaker 

BaseMaker(base: Base | None = None)

Bases: BaseMakerBase

Create a Base object and load repos

Initialize and configure the base object.

Methods:

  • create_repo –

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

  • disable_repo –

    Disable a repo by its id.

  • enable_repo –

    Enable a repo by its id.

  • enable_repos –

    Enable a list of repositories by their repoid.

  • enable_source_repos –

    Enable the corresponding -source repos of the currently enabled

  • fill_sack –

    Fill the sack and returns the dnf.Base object.

  • load_changelogs –

    Load changelog metadata

  • load_filelists –

    Load the filelists if they’re not already enabled by default

  • load_release_repos –

    Load the repositories from a fedrq.config.Release object

  • read_repofile –

    Load repositories from a repo file

  • read_system_repos –

    Load system repositories into the base object.

  • repolist –

    List available repoids.

  • set –

    Set configuration options. Must be called before reading repos.

  • set_var –

    Set substitutions (e.g. arch, basearch, releasever).

  • sets –

    Set options on the base object

Attributes:

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

backend property 

backend: BackendMod

Return the backend’s module

base instance-attribute 

base: Base = base or Base()

conf property 

conf: MainConf

Return the backend’s Config object

create_repo 

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

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

Parameters:

  • kwargs (Any, default: {} ) –

    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 
191
192
193
194
195
196
197
198
199
200
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._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 
157
158
159
160
161
162
163
164
165
166
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._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 
147
148
149
150
151
152
153
154
155
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._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 
139
140
141
142
143
144
145
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

Enable the corresponding -source repos of the currently enabled repositories

Source code in src/fedrq/backends/dnf/backend/__init__.py 
211
212
def enable_source_repos(self) -> None:
    self._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 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
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

Load changelog metadata

Parameters:

  • enable (bool, default: True ) –

    Whether to enable or disable filelists

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

load_filelists 

load_filelists(enable: bool = True) -> None

Load the filelists if they’re not already enabled by default

Parameters:

  • enable (bool, default: True ) –

    Whether to enable or disable filelists

Source code in src/fedrq/backends/dnf/backend/__init__.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def load_filelists(self, enable: bool = True) -> None:
    # Old versions of dnf always load filelists
    try:
        types: list[str] = (
            self.conf.optional_metadata_types
        )  # pyright: ignore[reportAssignmentType]
    except AttributeError:
        return
    if enable:
        types.append("filelists")
        return
    while "filelists" in types:
        types.remove("filelists")

load_release_repos 

load_release_repos(
    release: Release, set_releasever: bool = True
) -> None

Load the repositories from a fedrq.config.Release object

Parameters:

  • release (Release) –

    Release object

  • set_releasever (bool, default: True ) –

    Whether to set the $releasever based on the release or just leave it alone

Source code in src/fedrq/backends/base/__init__.py 
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
def load_release_repos(self, release: Release, set_releasever: bool = True) -> None:
    """
    Load the repositories from a fedrq.config.Release object

    Args:
        release:
            [`Release`][fedrq.config.Release] object
        set_releasever:
            Whether to set the `$releasever` based on the release or just
            leave it alone
    """
    if set_releasever:
        self.set_var("releasever", release.version)
    if release.release_config.system_repos:
        self.read_system_repos(
            disable=not (
                release.repo_name in {"@base", "base"}
                and release.release_config.append_system_repos
            )
        )
    for path in release.release_config.full_def_paths:
        with importlib.resources.as_file(path) as fp:
            LOG.debug("Reading %s", fp)
            self._read_repofile_new(fp)
    release.repog.load(self, release.config, release)

read_repofile 

read_repofile(file: StrPath) -> None

Load repositories from a repo file

Source code in src/fedrq/backends/dnf/backend/__init__.py 
168
169
170
171
172
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._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 
127
128
129
130
131
132
133
134
135
136
137
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._repos.iter_enabled():
        repo.disable()

repolist 

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

List available repoids.

Parameters:

  • enabled (bool | None, default: None ) –

    Only show enabled repositories

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

set 

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

Set configuration options. Must be called before reading repos.

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

set_var 

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

Set substitutions (e.g. arch, basearch, releasever). Needs to be called before reading repos.

Source code in src/fedrq/backends/dnf/backend/__init__.py 
86
87
88
89
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)

sets 

sets(
    conf: dict[str, Any], substitutions: dict[str, Any]
) -> None

Set options on the base object

Parameters:

  • conf (dict[str, Any]) –

    A dict of configuration options. Call self.set() for each k-v pair.

  • substitutions (dict[str, Any]) –

    A dict of substitutions/vars options. Call self.set_var() for each k-v pair.

Source code in src/fedrq/backends/base/__init__.py 
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
def sets(self, conf: dict[str, Any], substitutions: dict[str, Any]) -> None:
    """
    Set options on the base object

    Args:
        conf:
            A dict of configuration options. Call self.set() for each k-v
            pair.
        substitutions:
            A dict of substitutions/vars options. Call self.set_var() for
            each k-v pair.
    """
    for opt in conf.items():
        self.set(*opt)
    for opt in substitutions.items():
        self.set_var(*opt)

NEVRAForms

Bases: int, Enum

Enum of NEVRAForms to pass to Repoquery.resolve_pkg_specs

Attributes:

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: Base)

Bases: RepoqueryBase[PackageCompat, PackageQueryCompat[PackageCompat]]

Helpers to query a repository. Provides a unified repoquery interface for different backends.

Methods:

  • arch_filter –

    Filter a query’s architectures and return it.

  • arch_filterm –

    Filter a query’s architectures in place and return it.

  • get_package –

    Return the latest Package that matches the ‘name’ and ‘arch’.

  • get_subpackages –

    Return a PackageQuery containing the binary RPMS/subpackages produced

  • query –

    Return an inital PackageQuery that’s filtered with **kwargs.

  • resolve_pkg_specs –

    Resolve pkg specs.

Attributes:

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

backend property 

backend: BackendMod

Return fedrq.backends.dnf.backend module

base instance-attribute 

base: Base = base

base_arches property 

base_arches: set[str]

Return a set of the system’s arch and basearch.

arch_filter 

arch_filter(
    query: _PackageQueryT,
    arch: str | Iterable[str] | None = None,
) -> _PackageQueryT

Filter a query’s architectures and return it. It includes a little more functionality than query.filter(arch=…).

  • When arch is None, the query is left untouched.
  • If arch equals ‘notsrc’, all src and multilib packages are excluded.
  • If arch equals ‘arched’, all noarch, multilib, and source packages are excluded.
  • Otherwise, arch is passed to query.filterm(arch=…) and no other validation is preformed.
Source code in src/fedrq/backends/base/__init__.py 
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
def arch_filter(
    self: RepoqueryBase[PackageCompat, _PackageQueryT],
    query: _PackageQueryT,
    arch: str | Iterable[str] | None = None,
) -> _PackageQueryT:
    """
    Filter a query's architectures and return it.
    It includes a little more functionality than query.filter(arch=...).

    - When arch is None, the query is left untouched.
    - If arch equals 'notsrc', all src and multilib packages are
      excluded.
    - If arch equals 'arched', all noarch, multilib, and source
      packages are excluded.
    - Otherwise, arch is passed to query.filterm(arch=...) and no other
      validation is preformed.
    """
    if not arch:
        return query
    if arch == "notsrc":
        return query.filter(arch=(*self.base_arches, "noarch"))  # type: ignore
    if arch == "arched":
        return query.filter(arch=list(self.base_arches))
    return query.filter(arch=arch)

arch_filterm 

arch_filterm(
    query: _PackageQueryT,
    arch: str | Iterable[str] | None = None,
) -> _PackageQueryT

Filter a query’s architectures in place and return it. It includes a little more functionality than query.filterm(arch=…).

  • When arch is None, the query is left untouched.
  • If arch equals ‘notsrc’, all src and multilib packages are excluded.
  • If arch equals ‘arched’, all noarch, multilib, and source packages are excluded.
  • Otherwise, arch is passed to query.filterm(arch=…) and no other validation is preformed.
Source code in src/fedrq/backends/base/__init__.py 
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
def arch_filterm(
    self: RepoqueryBase[PackageCompat, _PackageQueryT],
    query: _PackageQueryT,
    arch: str | Iterable[str] | None = None,
) -> _PackageQueryT:
    """
    Filter a query's architectures in place and return it.
    It includes a little more functionality than query.filterm(arch=...).

    - When arch is None, the query is left untouched.
    - If arch equals 'notsrc', all src and multilib packages are
      excluded.
    - If arch equals 'arched', all noarch, multilib, and source
      packages are excluded.
    - Otherwise, arch is passed to query.filterm(arch=...) and no other
      validation is preformed.
    """
    if not arch:
        return query
    if arch == "notsrc":
        return query.filterm(arch=(*self.base_arches, "noarch"))  # type: ignore
    elif arch == "arched":
        return query.filterm(arch=self.base.conf.basearch)
    else:
        return query.filterm(arch=arch)

get_package 

get_package(
    name: str, arch: str | Iterable[str] | None = None
) -> _PackageT_co

Return the latest Package that matches the ‘name’ and ‘arch’. A ValueError is raised when no matches are found.

Source code in src/fedrq/backends/base/__init__.py 
620
621
622
623
624
625
626
627
628
629
630
631
632
def get_package(
    self,
    name: str,
    arch: str | Iterable[str] | None = None,
) -> _PackageT_co:
    """
    Return the latest Package that matches the 'name' and 'arch'.
    A ValueError is raised when no matches are found.
    """
    query = self.query(name=name, arch=arch, latest=1)
    if len(query) < 1:
        raise ValueError(f"Zero packages found for {name} on {arch}")
    return cast("_PackageT_co", next(iter(query)))

get_subpackages 

get_subpackages(
    packages: Iterable[_PackageT], **kwargs
) -> _PackageQueryT

Return a PackageQuery containing the binary RPMS/subpackages produced by {packages}.

Parameters:

  • packages (Iterable[_PackageT]) –

    An interable of PackageCompat containing source packages

Source code in src/fedrq/backends/base/__init__.py 
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
def get_subpackages(
    self: RepoqueryBase[_PackageT, _PackageQueryT],
    packages: Iterable[_PackageT],
    **kwargs,
) -> _PackageQueryT:
    """
    Return a PackageQuery containing the binary RPMS/subpackages produced
    by {packages}.

    Args:
        packages:
            An interable of `PackageCompat` containing source packages
    """
    arch = kwargs.get("arch")
    if arch == "src":
        raise ValueError("{arch} cannot be 'src'")
    elif not arch:
        kwargs.setdefault("arch__neq", "src")
    if val := kwargs.pop("sourcerpm", None):
        warn(f"Removing invalid kwarg: 'sourcerpm={val}")

    for package in packages:
        if package.arch != "src":
            raise ValueError(f"{package} must be a source package.")

    sourcerpms = [
        f"{package.name}-{package.version}-{package.release}.src.rpm"
        for package in packages
    ]
    query = self.query(sourcerpm=sourcerpms, **kwargs)
    return query

query 

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

Return an inital PackageQuery that’s filtered with **kwargs. Further filtering can be applied with the PackageQuery’s filter and filterm methods.

Source code in src/fedrq/backends/base/__init__.py 
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
def query(
    self,
    *,
    arch: str | Iterable[str] | None = None,
    **kwargs,
) -> _PackageQueryT_co:
    """
    Return an inital PackageQuery that's filtered with **kwargs.
    Further filtering can be applied with the PackageQuery's filter and
    filterm methods.
    """
    if kwargs.get("latest") is None:
        kwargs.pop("latest", None)
    query = self._query()
    query.filterm(**kwargs)
    self.arch_filterm(query, arch)
    return query

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 | int] | None = None
)

Resolve pkg specs. See https://dnf.readthedocs.io/en/latest/command_ref.html?highlight=spec#specifying-packages or https://dnf5.readthedocs.io/en/latest/misc/specs.7.html for valid forms.

Parameters:

  • specs (Collection[str]) –

    Package specs to resolve.

  • resolve (bool, default: False ) –

    Whether to resolve file paths or virtual Provides in addition to package specs

  • latest (int | None, default: None ) –

    Limit packages with the same name and arch.

  • with_src (bool, default: True ) –

    Whether to consider .src packages when resolving specs

Source code in src/fedrq/backends/dnf/backend/__init__.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
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 | int] | 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: Any) -> Iterator[ChangelogEntry]

Given a Package object, return an iterator of ChangelogEntry objects.

Source code in src/fedrq/backends/dnf/backend/__init__.py 
292
293
294
295
296
297
298
299
def get_changelogs(package: t.Any) -> Iterator[ChangelogEntry]:
    """
    Given a Package object, return an iterator of ChangelogEntry objects.
    """
    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 
284
285
286
287
288
289
@cache
def get_releasever():
    """
    Return the system releasever
    """
    return dnf.rpm.detect_releasever("/")

See also