ahriman.core package

Subpackages

Submodules

ahriman.core.exceptions module

exception BuildError(package_base: str, stderr: str | None = None)

Bases: RuntimeError

base exception for failed builds

Parameters:

  • package_base (str) – package base raised exception

  • stderr (str | None, optional) – stderr of the process if available (Default value = None)

classmethod from_process(package_base: str) → Callable[[int, list[str], str, str], Self]

generate exception callable from process error

Parameters:

package_base (str) – package base raised exception

Returns:

exception generator to be passed to subprocess utils

Return type:

Callable[[int, list[str], str, str], Self]

exception CalledProcessError(status_code: int, process: list[str], stderr: str)

Bases: CalledProcessError

like subprocess.CalledProcessError, but better

Parameters:

  • status_code (int) – process return code

  • process (list[str]) – process argument list

  • stderr (str) – stderr of the process

exception DuplicateRunError

Bases: RuntimeError

exception which will be raised if there is another application instance

exception ExitCode

Bases: RuntimeError

special exception which has to be thrown to return non-zero status without error message

exception ExtensionError

Bases: RuntimeError

exception being raised by trigger load in case of errors

exception GPGError(details: str)

Bases: RuntimeError

PGP/GPG related exception

Parameters:

details (str) – details of the exception

exception GitRemoteError

Bases: RuntimeError

git remote exception

exception InitializeError(details: str)

Bases: RuntimeError

base service initialization exception

Parameters:

details (str) – details of the exception

exception MigrationError(details: str)

Bases: RuntimeError

exception which will be raised on migration error

Parameters:

details (str) – error details

exception MissingArchitectureError(command: str)

Bases: ValueError

exception which will be raised if architecture is required, but missing

Parameters:

command (str) – command name which throws exception

exception MultipleArchitecturesError(command: str, repositories: list[RepositoryId] | None = None)

Bases: ValueError

exception which will be raised if multiple architectures are not supported by the handler

Parameters:

  • command (str) – command name which throws exception

  • repositories (list[RepositoryId] | None, optional) – found repository list (Default value = None)

exception OptionError(value: Any)

Bases: ValueError

exception which will be raised on configuration errors

Parameters:

value (Any) – option value

exception PackageInfoError(details: Any)

Bases: RuntimeError

exception which will be raised on package load errors

Parameters:

details (Any) – error details

exception PacmanError(details: Any)

Bases: RuntimeError

exception in case of pacman operation errors

Parameters:

details (Any) – error details

exception PartitionError(count: int)

Bases: RuntimeError

exception raised during packages partition actions

Parameters:

count (int) – count of partitions

exception PasswordError(details: Any)

Bases: ValueError

exception which will be raised in case of password related errors

Parameters:

details (Any)

exception PkgbuildGeneratorError

Bases: RuntimeError

exception class for support type triggers

exception PkgbuildParserError(reason: str, source: Any = None)

Bases: ValueError

exception raises in case of PKGBUILD parser errors

Parameters:

  • reason (str) – parser error reason

  • source (Any, optional) – source line if available (Default value = None)

exception ReportError

Bases: RuntimeError

report generation exception

exception SynchronizationError

Bases: RuntimeError

remote synchronization exception

exception UnknownPackageError(package_base: str)

Bases: ValueError

exception for status watcher which will be thrown on unknown package

Parameters:

package_base (str) – package base name

exception UnsafeRunError(current_uid: int, root_uid: int)

Bases: RuntimeError

exception which will be raised in case if user is not owner of repository

Parameters:

  • current_uid (int) – current user ID

  • root_uid (int) – ID of the owner of root directory

ahriman.core.module_loader module

implementations(root_module: ModuleType, base_class: type[T]) → Iterator[type[T]]

extract implementations of the base_class from the module

Parameters:

  • root_module (ModuleType) – root module

  • base_class (type[T]) – base class type

Yields:

type[T] – found implementations

ahriman.core.spawn module

class Spawn(args_parser: ArgumentParser, command_arguments: list[str])

Bases: Thread, LazyLogging

helper to spawn external ahriman process MUST NOT be used directly, the only one usage allowed is to spawn process from web services

active

map of active child processes required to avoid zombies

Type:

dict[str, Process]

command_arguments

base command line arguments

Type:

list[str]

queue

multiprocessing queue to read updates from processes

Type:

Queue[ProcessStatus | None]

Parameters:

  • args_parser (argparse.ArgumentParser) – command line parser for the application

  • command_arguments (list[str]) – base command line arguments

static boolean_action_argument(name: str, value: bool) → str

convert option of given name with value to boolean action argument

Parameters:

  • name (str) – command line argument name

  • value (bool) – command line argument value

Returns:

if value is True, then returns positive flag and negative otherwise

Return type:

str

static process(callback: Callable[[argparse.Namespace, RepositoryId], bool], args: argparse.Namespace, repository_id: RepositoryId, process_id: str, queue: Queue[ProcessStatus | None]) → None

helper to run external process

Parameters:
has_process(process_id: str) → bool

check if given process is alive

Parameters:

process_id (str) – process id to be checked as returned by _spawn_process()

Returns:

True in case if process still counts as active and False otherwise

Return type:

bool

key_import(key: str, server: str | None) → str

import key to service cache

Parameters:

  • key (str) – key to import

  • server (str | None) – PGP key server

Returns:

spawned process identifier

Return type:

str

packages_add(repository_id: RepositoryId, packages: Iterable[str], username: str | None, *, patches: list[PkgbuildPatch], now: bool, increment: bool, refresh: bool) → str

add packages

Parameters:

  • repository_id (RepositoryId) – repository unique identifier

  • packages (Iterable[str]) – packages list to add

  • username (str | None) – optional override of username for build process

  • patches (list[PkgbuildPatch]) – list of patches to be passed

  • now (bool) – build packages now

  • increment (bool) – increment pkgrel on conflict

  • refresh (bool) – refresh pacman database before process

Returns:

spawned process identifier

Return type:

str

packages_rebuild(repository_id: RepositoryId, depends_on: str, username: str | None, *, increment: bool) → str

rebuild packages which depend on the specified package

Parameters:

  • repository_id (RepositoryId) – repository unique identifier

  • depends_on (str) – packages dependency

  • username (str | None) – optional override of username for build process

  • increment (bool) – increment pkgrel on conflict

Returns:

spawned process identifier

Return type:

str

packages_remove(repository_id: RepositoryId, packages: Iterable[str]) → str

remove packages

Parameters:

  • repository_id (RepositoryId) – repository unique identifier

  • packages (Iterable[str]) – packages list to remove

Returns:

spawned process identifier

Return type:

str

packages_update(repository_id: RepositoryId, username: str | None, *, aur: bool, local: bool, manual: bool, increment: bool, refresh: bool) → str

run full repository update

Parameters:

  • repository_id (RepositoryId) – repository unique identifier

  • username (str | None) – optional override of username for build process

  • aur (bool) – check for aur updates

  • local (bool) – check for local packages updates

  • manual (bool) – check for manual packages

  • increment (bool) – increment pkgrel on conflict

  • refresh (bool) – refresh pacman database before process

Returns:

spawned process identifier

Return type:

str

run() → None

thread run method

stop() → None

gracefully terminate thread

ahriman.core.tree module

class Leaf(package: Package)

Bases: object

tree leaf implementation

dependencies

list of package dependencies

Type:

set[str]

items

list of packages in this leaf including provides

Type:

list[str]

package

leaf package properties

Type:

Package

Parameters:

package (Package) – package properties

is_dependency(packages: Iterable[Leaf]) → bool

check if the package is dependency of any other package from list or not

Parameters:

packages (Iterable[Leaf]) – list of known leaves

Returns:

True in case if package is dependency of others and False otherwise

Return type:

bool

is_root(packages: Iterable[Leaf]) → bool

check if package depends on any other package from list of not

Parameters:

packages (Iterable[Leaf]) – list of known leaves

Returns:

True if any of packages is dependency of the leaf, False otherwise

Return type:

bool

class Tree(leaves: list[Leaf])

Bases: object

dependency tree implementation

leaves

list of tree leaves

Type:

list[Leaf]

Examples

The most important feature here is to generate tree levels one by one which can be achieved by using class method:

>>> from ahriman.core.configuration import Configuration
>>> from ahriman.core.database import SQLite
>>> from ahriman.core.repository import Repository
>>> from ahriman.models.repository_id import RepositoryId
>>>
>>> configuration = Configuration()
>>> database = SQLite.load(configuration)
>>> repository = Repository.load(RepositoryId("x86_64", "aur"), configuration, database, report=True)
>>> packages = repository.packages()
>>>
>>> tree = Tree.resolve(packages)
>>> for tree_level in tree:
>>>     for package in tree_level:
>>>         print(package.base)
>>>     print()

The direct constructor call is also possible but requires tree leaves to be instantioned in advance, e.g.:

>>> leaves = [Leaf(package) for package in packages]
>>> tree = Tree(leaves)
Parameters:

leaves (list[Leaf]) – leaves to build the tree

static balance(partitions: list[list[Leaf]]) → list[list[Leaf]]

balance partitions. This method tries to find the longest and the shortest lists and move free leaves between them if possible. In case if there are no free packages (i.e. the ones which don’t depend on any other in partition and are not dependency of any), it will drop it as it is. This method is guaranteed to produce the same unsorted sequences for same unsorted input

Parameters:

partitions (list[list[Leaf]]) – source unbalanced partitions

Returns:

balanced partitions

Return type:

list[list[Leaf]]

static partition(packages: Iterable[Package], *, count: int) → list[list[Package]]

partition tree into independent chunks of more or less equal amount of packages. The packages in produced partitions don’t depend on any package from other partitions

Parameters:

  • packages (Iterable[Package]) – packages list

  • count (int) – maximal amount of partitions

Returns:

list of packages lists based on their dependencies. The amount of elements in each sublist is less or equal to count

Return type:

list[list[Package]]

Raises:

PartitionError – in case if it is impossible to divide tree by specified amount of partitions

static resolve(packages: Iterable[Package]) → list[list[Package]]

resolve dependency tree

Parameters:

packages (Iterable[Package]) – packages list

Returns:

list of packages lists based on their dependencies

Return type:

list[list[Package]]

static sort(leaves: list[list[Leaf]]) → list[list[Package]]

sort given list of leaves by package base

Parameters:

leaves (list[list[Leaf]]) – leaves to sort

Returns:

sorted list of packages on each level

Return type:

list[list[Package]]

levels() → list[list[Package]]

get build levels starting from the packages which do not require any other package to build

Returns:

sorted list of packages lists based on their dependencies

Return type:

list[list[Package]]

partitions(*, count: int) → list[list[Package]]

partition tree into (more or less) equal chunks of packages which don’t depend on each other

Parameters:

count (int) – maximal amount of partitions

Returns:

sorted list of packages partitions

Return type:

list[list[Package]]

ahriman.core.types module

class Comparable(*args, **kwargs)

Bases: Protocol

class which supports __lt__() operation`

class HasBool(*args, **kwargs)

Bases: Protocol

class which defines bool() method

class HasLength(*args, **kwargs)

Bases: Protocol

class which defines len() method

ahriman.core.util module

ahriman.core.utils module

atomic_move(src: Path, dst: Path) → None

move file from source location to destination. This method uses lock and shutil.move() to ensure that file will be copied (if not rename) atomically. This method blocks execution until lock is available

Parameters:

  • src (Path) – path to the source file

  • dst (Path) – path to the destination

Examples

This method is a drop-in replacement for shutil.move() (except it doesn’t allow to override copy method) which first locking destination file. To use it simply call method with arguments:

>>> atomic_move(src, dst)
check_output(*args: str, exception: Exception | Callable[[int, list[str], str, str], Exception] | None = None, cwd: Path | None = None, input_data: str | None = None, logger: Logger | None = None, user: int | None = None, environment: dict[str, str] | None = None) → str

subprocess wrapper

Parameters:

  • *args (str) – command line arguments

  • exception (Exception | Callable[[int, list[str], str, str]] | None, optional) – exception which has to be raised instead of default subprocess exception. If callable us is supplied, the subprocess.CalledProcessError arguments will be passed (Default value = None)

  • cwd (Path | None, optional) – current working directory (Default value = None)

  • input_data (str | None, optional) – data which will be written to command stdin (Default value = None)

  • logger (logging.Logger | None, optional) – logger to log command result if required (Default value = None)

  • user (int | None, optional) – run process as specified user (Default value = None)

  • environment (dict[str, str] | None, optional) – optional environment variables if any (Default value = None)

Returns:

command output

Return type:

str

Raises:

CalledProcessError – if subprocess ended with status code different from 0 and no exception supplied

Examples

Simply call the function:

>>> check_output("echo", "hello world")

The more complicated calls which include result logging and input data are also possible:

>>> import logging
>>>
>>> logger = logging.getLogger()
>>> check_output("python", "-c", "greeting = input('say hello: '); print(); print(greeting)",
>>>              input_data="hello world", logger=logger)

An additional argument exception can be supplied in order to override the default exception:

>>> check_output("false", exception=RuntimeError("An exception occurred"))
check_user(root: Path, *, unsafe: bool) → None

check if current user is the owner of the root

Parameters:
Raises:

UnsafeRunError – if root uid differs from current uid and check is enabled

Examples

Simply run function with arguments:

>>> check_user(root, unsafe=False)
dataclass_view(instance: Any) → dict[str, Any]

convert dataclass instance to json object

Parameters:

instance (Any) – dataclass instance

Returns:

json representation of the dataclass with empty field removed

Return type:

dict[str, Any]

enum_values(enum: type[Enum]) → list[str]

generate list of enumeration values from the source

Parameters:

enum (type[Enum]) – source enumeration class

Returns:

available enumeration values as string

Return type:

list[str]

extract_user() → str | None

extract user from system environment

Returns:

SUDO_USER in case if set and USER otherwise. It can return None in case if environment has been cleared before application start

Return type:

str | None

filelock(path: Path) → Iterator[UnixFileLock]

wrapper around filelock.FileLock, which also removes locks afterward

Parameters:

path (Path) – path to lock on. The lock file will be created as .{path.name}.lock

Yields:

FileLock – acquired file lock instance

filter_json(source: T, known_fields: Iterable[str] | None = None) → T

recursively filter json object removing None values and optionally filtering by known fields

Parameters:

  • source (T) – raw json object (dict, list, or scalar)

  • known_fields (Iterable[str] | None, optional) – list of fields which have to be known for the target object (Default value = None)

Returns:

json without None values

Return type:

T

Examples

This wrapper is mainly used for the dataclasses, thus the flow must be something like this:

>>> from dataclasses import fields
>>> from ahriman.models.package import Package
>>>
>>> known_fields = [pair.name for pair in fields(Package)]
>>> properties = filter_json(dump, known_fields)
>>> package = Package(**properties)
full_version(epoch: str | int | None, pkgver: str, pkgrel: str) → str

generate full version from components

Parameters:

  • epoch (str | int | None) – package epoch if any

  • pkgver (str) – package version

  • pkgrel (str) – package release version (arch linux specific)

Returns:

generated version

Return type:

str

list_flatmap(source: Iterable[T], extractor: Callable[[T], Iterable[R]]) → list[R]

extract elements from list of lists, flatten them and apply extractor

Parameters:

  • source (Iterable[T]) – source list

  • extractor (Callable[[T], Iterable[R]]) – property extractor

Returns:

combined list of unique entries in properties list

Return type:

list[R]

minmax(source: Iterable[T], *, key: Callable[[T], Any] | None = None) → tuple[T, T]

get min and max value from iterable

Parameters:

  • source (Iterable[T]) – source list to find min and max values

  • key (Callable[[T], Any] | None, optional) – key to sort (Default value = None)

Returns:

min and max values for sequence

Return type:

tuple[T, T]

owner(path: Path) → tuple[int, int]

retrieve owner information by path

Parameters:

path (Path) – path for which extract ids

Returns:

owner user and group ids of the directory

Return type:

tuple[int, int]

package_like(filename: Path) → bool

check if file looks like package

Parameters:

filename (Path) – name of file to check

Returns:

True in case if name contains .pkg. and not signature, False otherwise

Return type:

bool

parse_version(version: str) → tuple[str | None, str, str]

parse version and returns its components

Parameters:

version (str) – full version string

Returns:

epoch if any, pkgver and pkgrel variables

Return type:

tuple[str | None, str, str]

partition(source: Iterable[T], predicate: Callable[[T], bool]) → tuple[list[T], list[T]]

partition list into two based on predicate, based on https://docs.python.org/dev/library/itertools.html#itertools-recipes

Parameters:

  • source (Iterable[T]) – source list to be partitioned

  • predicate (Callable[[T], bool]) – filter function

Returns:

two lists, first is which predicate is True, second is False

Return type:

tuple[list[T], list[T]]

pretty_datetime(timestamp: datetime | float | int | None) → str

convert datetime object to string

Parameters:

timestamp (datetime.datetime | float | int | None) – datetime to convert

Returns:

pretty printable datetime as string

Return type:

str

pretty_interval(interval: int) → str

convert time interval to string

Parameters:

interval (int) – time interval in seconds

Returns:

pretty printable interval as string

Return type:

str

pretty_size(size: float | None, level: int = 0) → str

convert size to string

Parameters:

  • size (float | None) – size to convert

  • level (int, optional) – represents current units, 0 is B, 1 is KiB, etc. (Default value = 0)

Returns:

pretty printable size as string

Return type:

str

Raises:

OptionError – if size is more than 1TiB

safe_filename(source: str) → str

convert source string to its safe representation

Parameters:

source (str) – string to convert

Returns:

result string in which all unsafe characters are replaced by dash

Return type:

str

srcinfo_property(key: str, srcinfo: Mapping[str, Any], package_srcinfo: Mapping[str, Any], *, default: Any = None) → Any

extract property from SRCINFO. This method extracts property from package if this property is presented in srcinfo. Otherwise, it looks for the same property in root srcinfo. If none found, the default value will be returned

Parameters:

  • key (str) – key to extract

  • srcinfo (Mapping[str, Any]) – root structure of SRCINFO

  • package_srcinfo (Mapping[str, Any]) – package specific SRCINFO

  • default (Any, optional) – the default value for the specified key (Default value = None)

Returns:

extracted value from SRCINFO

Return type:

Any

srcinfo_property_list(key: str, srcinfo: Mapping[str, Any], package_srcinfo: Mapping[str, Any], *, architecture: str | None = None) → list[Any]

extract list property from SRCINFO. Unlike srcinfo_property() it supposes that default return value is always empty list. If architecture is supplied, then it will try to lookup for architecture specific values and will append it at the end of result

Parameters:

  • key (str) – key to extract

  • srcinfo (Mapping[str, Any]) – root structure of SRCINFO

  • package_srcinfo (Mapping[str, Any]) – package specific SRCINFO

  • architecture (str | None, optional) – package architecture if set (Default value = None)

Returns:

list of extracted properties from SRCINFO

Return type:

list[Any]

create symlink with relative path to the target directory

Parameters:

  • symlink (Path) – path to symlink to create

  • source (Path) – source file to be symlinked

trim_package(package_name: str) → str

remove version bound and description from package name. Pacman allows to specify version bound (=, <=, >= etc.) for packages in dependencies and also allows to specify description (via :); this function removes trailing parts and return exact package name

Parameters:

package_name (str) – source package name

Returns:

package name without description or version bound

Return type:

str

utcnow() → datetime

get current time

Returns:

current time in UTC

Return type:

datetime.datetime

walk(directory_path: Path) → Iterator[Path]

list all file paths in given directory

Parameters:

directory_path (Path) – root directory path

Yields:

Path – all found files in given directory with full path

Examples

Wrapper around pathlib.Path.walk(), which yields only files instead:

>>> from pathlib import Path
>>>
>>> for file_path in walk(Path.cwd()):
>>>     print(file_path)

Module contents