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

default constructor

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

default constructor

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

default constructor

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 GitRemoteError

Bases: RuntimeError

git remote exception

default constructor

exception InitializeError(details: str)

Bases: RuntimeError

base service initialization exception

default constructor

Parameters:

details (str) – details of the exception

exception MigrationError(details: str)

Bases: RuntimeError

exception which will be raised on migration error

default constructor

Parameters:

details (str) – error details

exception MissingArchitectureError(command: str)

Bases: ValueError

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

default constructor

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

default constructor

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

default constructor

Parameters:

value (Any) – option value

exception PackageInfoError(details: Any)

Bases: RuntimeError

exception which will be raised on package load errors

default constructor

Parameters:

details (Any) – error details

exception PacmanError(details: Any)

Bases: RuntimeError

exception in case of pacman operation errors

default constructor

Parameters:

details (Any) – error details

exception PartitionError(count: int)

Bases: RuntimeError

exception raised during packages partition actions

default constructor

Parameters:

count (int) – count of partitions

exception PasswordError(details: Any)

Bases: ValueError

exception which will be raised in case of password related errors

default constructor

Parameters:

details (Any) –

exception PathError(path: Path, root: Path)

Bases: ValueError

exception which will be raised on path which is not belong to root directory

default constructor

Parameters:

  • path (Path) – path which raised an exception

  • root (Path) – repository root (i.e. ahriman home)

exception PkgbuildGeneratorError

Bases: RuntimeError

exception class for support type triggers

default constructor

exception ReportError

Bases: RuntimeError

report generation exception

default constructor

exception SynchronizationError

Bases: RuntimeError

remote synchronization exception

default constructor

exception UnknownPackageError(package_base: str)

Bases: ValueError

exception for status watcher which will be thrown on unknown package

default constructor

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

default constructor

Parameters:

  • current_uid (int) – current user ID

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

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]

default constructor

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

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

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:
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]

package

leaf package properties

Type:

Package

default constructor

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

property items: Iterable[str]

extract all packages from the leaf

Returns:

packages containing in this leaf

Return type:

Iterable[str]

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-clone"), 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)

default constructor

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]]

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]]

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

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]]

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]]

ahriman.core.util module

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(paths: RepositoryPaths, *, unsafe: bool) → None

check if current user is the owner of the root

Parameters:

  • paths (RepositoryPaths) – repository paths object

  • unsafe (bool) – if set no user check will be performed before path creation

Raises:

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

Examples

Simply run function with arguments:

>>> check_user(paths, 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

Return type:

str | None

cleared before application start

filter_json(source: dict[str, Any], known_fields: Iterable[str]) → dict[str, Any]

filter json object by fields used for json-to-object conversion

Parameters:

  • source (dict[str, Any]) – raw json object

  • known_fields (Iterable[str]) – list of fields which have to be known for the target object

Returns:

json object without unknown and empty fields

Return type:

dict[str, Any]

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

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]

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_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: dict[str, Any], package_srcinfo: dict[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 (dict[str, Any]) – root structure of SRCINFO

  • package_srcinfo (dict[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: dict[str, Any], package_srcinfo: dict[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 (dict[str, Any]) – root structure of SRCINFO

  • package_srcinfo (dict[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]

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

unquote(source: str) → str

like shlex.quote(), but opposite

Parameters:

source (str) – source string to remove quotes

Returns:

string with quotes removed

Return type:

str

Raises:

ValueError – if no closing quotation

utcnow() → datetime

get current time

Returns:

current time in UTC

Return type:

datetime.datetime

walk(directory_path: Path) → Generator[Path, None, None]

list all file paths in given directory Credits to https://stackoverflow.com/a/64915960

Parameters:

directory_path (Path) – root directory path

Yields:

Path – all found files in given directory with full path

Examples

Since the pathlib module does not provide an alternative to os.walk(), this wrapper can be used instead:

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

Note, however, that unlike the original method, it does not yield directories.

Module contents