fedrq.config

This module houses code to load configuration from the filesystem and validate it.

CONFIG_DIRS module-attribute

CONFIG_DIRS = (home() / ".config/fedrq", Path("/etc/fedrq"))

DEFAULT_COPR_BASEURL module-attribute

DEFAULT_COPR_BASEURL = 'https://copr.fedoraproject.org'

DEFAULT_REPO_CLASS module-attribute

DEFAULT_REPO_CLASS = 'base'

LoadFilelists

Bases: StrEnum

always class-attribute instance-attribute

always = auto()

auto class-attribute instance-attribute

auto = auto()

never class-attribute instance-attribute

never = auto()

from_bool classmethod

from_bool(boolean: bool) -> LoadFilelists

Source code in src/fedrq/config.py

@classmethod
def from_bool(cls, /, boolean: bool) -> LoadFilelists:
    return cls.always if boolean else cls.never

RQConfig

Bases: BaseModel

backend class-attribute instance-attribute

backend: Optional[str] = get('FEDRQ_BACKEND')

backend_mod property

backend_mod: BackendMod

copr_baseurl class-attribute instance-attribute

copr_baseurl: str = DEFAULT_COPR_BASEURL

default_branch class-attribute instance-attribute

default_branch: str = get('FEDRQ_BRANCH', 'rawhide')

load_filelists class-attribute instance-attribute

load_filelists: LoadFilelists = auto

load_other_metadata class-attribute instance-attribute

load_other_metadata: Optional[bool] = None

release_names property

release_names: list[str]

releases instance-attribute

releases: dict[str, ReleaseConfig]

smartcache class-attribute instance-attribute

smartcache: Union[bool, Literal['always']] = True

Config

json_encoders class-attribute instance-attribute

json_encoders: dict[Any, Callable[[Any], str]] = {
    Pattern: lambda pattern: pattern,
    Path: lambda path: str(path),
}

validate_assignment class-attribute instance-attribute

validate_assignment = True

get_release

get_release(
    branch: str | None = None, repo_name: str | None = None
) -> Release

Source code in src/fedrq/config.py

def get_release(
    self, branch: str | None = None, repo_name: str | None = None
) -> Release:
    flog = mklog(__name__, "RQConfig", "get_releases")
    branch = branch or self.default_branch
    repo_name = repo_name or DEFAULT_REPO_CLASS
    pair = (branch, repo_name)
    for release in sorted(
        self.releases.values(), key=lambda r: r.name, reverse=True
    ):
        try:
            r = release.get_release(self, branch=branch, repo_name=repo_name)
        except ConfigError as exc:
            logger.debug(f"{release.name} does not match {pair}: {exc}")
        else:
            flog.debug("%s matches %s", release.name, pair)
            return r
    raise ConfigError(
        "{} does not much any of the configured releases: {}".format(
            pair, self.release_names
        )
    )

get_rq

get_rq(
    branch: str | None = None,
    repo: str | None = None,
    base_conf: dict[str, t.Any] | None = None,
    base_vars: dict[str, t.Any] | None = None,
) -> (
    _dnfRepoquery
    | _libdnf5RepoQuery
    | RepoqueryBase[PackageCompat]
)

Higher level interface that finds the Release object that mathces {branch} and {repo}, creates a (lib)dnf(5).base.Base session, and returns a Repoquery object.

Parameters:

Name	Type	Description	Default
branch	str | None	branch name	None
repo	str | None	repo class. defaults to ‘base’.	None
base_conf	dict[str, Any] | None	Base session configuration	None
base_vars	dict[str, Any] | None	Base session vars/substitutions (arch, basearch, releasever, etc.)	None

Source code in src/fedrq/config.py

def get_rq(
    self,
    branch: str | None = None,
    repo: str | None = None,
    base_conf: dict[str, t.Any] | None = None,
    base_vars: dict[str, t.Any] | None = None,
) -> _dnfRepoquery | _libdnf5RepoQuery | RepoqueryBase[PackageCompat]:
    """
    Higher level interface that finds the Release object that mathces
    {branch} and {repo}, creates a (lib)dnf(5).base.Base session, and
    returns a Repoquery object.

    Args:
        branch:
            branch name
        repo:
            repo class. defaults to 'base'.
        base_conf:
            Base session configuration
        base_vars:
            Base session vars/substitutions (arch, basearch, releasever,
                                             etc.)
    """
    release = self.get_release(branch, repo)
    return self.backend_mod.Repoquery(release.make_base(self, base_conf, base_vars))

Release

Release(
    config: RQConfig,
    release_config: ReleaseConfig,
    branch: str,
    repo_name: str = "base",
)

Encapsulates a ReleaseConfig with a specific version and repo name. This SHOULD NOT be instantiated directly. The init() has no stability promises. Use the RQConfig.get_release() factory instead.

Source code in src/fedrq/config.py

def __init__(
    self,
    config: RQConfig,
    release_config: ReleaseConfig,
    branch: str,
    repo_name: str = "base",
) -> None:
    self.config = config
    self.release_config = release_config
    if not self.release_config.is_match(branch):
        raise ConfigError(
            f"Branch {branch} does not match {self.release_config.name}"
        )
    self.branch = branch
    self.repo_name = repo_name
    self.repog: RepoG = self.get_repog(repo_name)

branch instance-attribute

branch = branch

config instance-attribute

config = config

copr_chroot_fmt property

copr_chroot_fmt: str | None

koschei_collection property

koschei_collection: str | None

release_config instance-attribute

release_config = release_config

repo_name instance-attribute

repo_name = repo_name

repog instance-attribute

repog: RepoG = get_repog(repo_name)

version property

version: str

get_repog

get_repog(key: str) -> RepoG

Source code in src/fedrq/config.py

def get_repog(self, key: str) -> RepoG:
    return self.release_config.repogs.get_repo(key)

make_base

make_base(
    config: RQConfig | None = None,
    base_conf: dict[str, t.Any] | None = None,
    base_vars: dict[str, t.Any] | None = None,
    base_maker: BaseMakerBase | None = None,
    fill_sack: bool = True,
) -> dnf.Base | libdnf5.base.Base

Parameters:

Name	Type	Description	Default
config	RQConfig | None	An RQConfig object. If this is not passed, self.config is used.	None
base_conf	dict[str, Any] | None	Base session configuration	None
base_vars	dict[str, Any] | None	Base session vars/substitutions (arch, basearch, releasever, etc.)	None
base_maker	BaseMakerBase | None	Existing BaseMaker object to configure. If base_maker is None, a new one will be created.	None
fill_sack	bool	Whether to fill the Base object’s package sack or just return the Base object after applying configuration.	True

Source code in src/fedrq/config.py

def make_base(
    self,
    config: RQConfig | None = None,
    base_conf: dict[str, t.Any] | None = None,
    base_vars: dict[str, t.Any] | None = None,
    base_maker: BaseMakerBase | None = None,
    fill_sack: bool = True,
) -> dnf.Base | libdnf5.base.Base:
    """
    Args:
        config:
            An RQConfig object. If this is not passed, `self.config` is used.
        base_conf:
            Base session configuration
        base_vars:
            Base session vars/substitutions (arch, basearch,
                                                       releasever, etc.)
        base_maker:
            Existing BaseMaker object to configure. If base_maker is None,
            a new one will be created.
        fill_sack:
            Whether to fill the Base object's package sack or just return
            the Base object after applying configuration.
    """
    if config is None:
        config = self.config
    base_conf = base_conf or {}
    base_vars = base_vars or {}
    releasever = config.backend_mod.get_releasever()
    if (
        "cachedir" not in base_conf
        and config.smartcache
        and (config.smartcache == "always" or self.version != releasever)
    ):
        logger.debug("Using smartcache")
        base_conf["cachedir"] = str(get_smartcache_basedir() / str(self.version))
    bm = base_maker or config.backend_mod.BaseMaker()
    bm.sets(base_conf, base_vars)
    bm.load_release_repos(self, "releasever" not in base_vars)
    if config.load_filelists:
        bm.load_filelists()
    if config.load_other_metadata is not None:
        bm.load_changelogs(config.load_other_metadata)
    return bm.fill_sack() if fill_sack else bm.base

ReleaseConfig

Bases: BaseModel

append_system_repos class-attribute instance-attribute

append_system_repos: bool = False

copr_chroot_fmt class-attribute instance-attribute

copr_chroot_fmt: Optional[str] = None

defpaths class-attribute instance-attribute

defpaths: set[str] = Field(default_factory=set)

defs instance-attribute

defs: dict[str, list[str]]

full_def_paths class-attribute instance-attribute

full_def_paths: list[Union[Traversable, Path]] = []

koschei_collection class-attribute instance-attribute

koschei_collection: Optional[str] = None

matcher instance-attribute

matcher: Pattern

name class-attribute instance-attribute

name: str = Field(exclude=True)

repo_aliases class-attribute instance-attribute

repo_aliases: dict[str, str] = {}

repo_dirs class-attribute instance-attribute

repo_dirs: list[Path] = Field(
    default_factory=lambda: [
        joinpath("repos") for directory in CONFIG_DIRS
    ]
)

repogs class-attribute instance-attribute

repogs: Repos = Field(DefaultRepoGs, exclude=True)

system_repos class-attribute instance-attribute

system_repos: bool = True

version class-attribute instance-attribute

version: Optional[str] = None

Config

arbitrary_types_allowed class-attribute instance-attribute

arbitrary_types_allowed = True

get_release

get_release(
    config: RQConfig, branch: str, repo_name: str = "base"
) -> Release

Source code in src/fedrq/config.py

def get_release(
    self, config: RQConfig, branch: str, repo_name: str = "base"
) -> Release:
    return Release(
        config=config, release_config=self, branch=branch, repo_name=repo_name
    )

is_match

is_match(val: str) -> bool

Source code in src/fedrq/config.py

def is_match(self, val: str) -> bool:
    return bool(re.fullmatch(self.matcher, val))

is_valid_repo

is_valid_repo(val: str) -> bool

Source code in src/fedrq/config.py

def is_valid_repo(self, val: str) -> bool:
    try:
        self.repogs.get_repo(val)
    except ConfigError:
        return False
    else:
        return True

get_config

get_config(**overrides: t.Any) -> RQConfig

Retrieve config files from CONFIG_DIRS and fedrq.data. Perform naive top-level merging of the ‘releases’ table.

Source code in src/fedrq/config.py

def get_config(**overrides: t.Any) -> RQConfig:
    """
    Retrieve config files from CONFIG_DIRS and fedrq.data.
    Perform naive top-level merging of the 'releases' table.
    """
    flog = mklog(__name__, "get_config")
    flog.debug(f"CONFIG_DIRS = {CONFIG_DIRS}")
    config: dict[str, t.Any] = {}
    all_files: list[list[t.Union[Traversable, Path]]] = [
        _get_files(importlib_resources.files("fedrq.data"), ".toml"),
        *(_get_files(p, ".toml") for p in reversed(CONFIG_DIRS)),
    ]
    flog.debug("all_files = %s", all_files)
    for path in itertools.chain.from_iterable(all_files):
        flog.debug("Loading config file: %s", path)
        with path.open("rb") as fp:
            data = tomllib.load(t.cast("t.BinaryIO", fp))
        merge_dict(data, config)
    merge_dict(overrides, config)
    config["releases"] = _get_releases(config["releases"])
    flog.debug("Final config: %s", config)
    return RQConfig(**config)

get_smartcache_basedir

get_smartcache_basedir() -> Path

Source code in src/fedrq/config.py

def get_smartcache_basedir() -> Path:
    basedir = Path(os.environ.get("XDG_CACHE_HOME", Path("~/.cache").expanduser()))
    return basedir.joinpath("fedrq").resolve()