ahriman.core package
Subpackages
- ahriman.core.alpm package
- ahriman.core.auth package
- ahriman.core.build_tools package
- Submodules
- ahriman.core.build_tools.sources module
Sources
Sources.DEFAULT_BRANCH
Sources.DEFAULT_COMMIT_AUTHOR
Sources.add()
Sources.changes()
Sources.commit()
Sources.diff()
Sources.extend_architectures()
Sources.fetch()
Sources.fetch_until()
Sources.has_changes()
Sources.has_remotes()
Sources.head()
Sources.init()
Sources.load()
Sources.move()
Sources.patch_apply()
Sources.patch_create()
Sources.push()
- ahriman.core.build_tools.task module
- Module contents
- ahriman.core.configuration package
- Submodules
- ahriman.core.configuration.configuration module
Configuration
Configuration.ARCHITECTURE_SPECIFIC_SECTIONS
Configuration.SYSTEM_CONFIGURATION_PATH
Configuration.includes
Configuration.path
Configuration.repository_id
Configuration.check_loaded()
Configuration.dump()
Configuration.from_path()
Configuration.gettype()
Configuration.load()
Configuration.load_includes()
Configuration.merge_sections()
Configuration.override_sections()
Configuration.reload()
Configuration.section_name()
Configuration.set_option()
Configuration.architecture
Configuration.include
Configuration.logging_path
Configuration.repository_name
Configuration.repository_paths
- ahriman.core.configuration.schema module
- ahriman.core.configuration.shell_interpolator module
- ahriman.core.configuration.validator module
- Module contents
- ahriman.core.database package
- Subpackages
- ahriman.core.database.migrations package
- Submodules
- ahriman.core.database.migrations.m000_initial module
- ahriman.core.database.migrations.m001_package_source module
- ahriman.core.database.migrations.m002_user_access module
- ahriman.core.database.migrations.m003_patch_variables module
- ahriman.core.database.migrations.m004_logs module
- ahriman.core.database.migrations.m005_make_opt_depends module
- ahriman.core.database.migrations.m006_packages_architecture_required module
- ahriman.core.database.migrations.m007_check_depends module
- ahriman.core.database.migrations.m008_packagers module
- ahriman.core.database.migrations.m009_local_source module
- ahriman.core.database.migrations.m010_version_based_logs_removal module
- ahriman.core.database.migrations.m011_repository_name module
- ahriman.core.database.migrations.m012_last_commit_sha module
- Module contents
- ahriman.core.database.operations package
- Submodules
- ahriman.core.database.operations.auth_operations module
- ahriman.core.database.operations.build_operations module
- ahriman.core.database.operations.changes_operations module
- ahriman.core.database.operations.logs_operations module
- ahriman.core.database.operations.operations module
- ahriman.core.database.operations.package_operations module
- ahriman.core.database.operations.patch_operations module
- Module contents
- ahriman.core.database.migrations package
- Submodules
- ahriman.core.database.sqlite module
- Module contents
- Subpackages
- ahriman.core.distributed package
- ahriman.core.formatters package
- Submodules
- ahriman.core.formatters.aur_printer module
- ahriman.core.formatters.build_printer module
- ahriman.core.formatters.changes_printer module
- ahriman.core.formatters.configuration_paths_printer module
- ahriman.core.formatters.configuration_printer module
- ahriman.core.formatters.package_printer module
- ahriman.core.formatters.patch_printer module
- ahriman.core.formatters.printer module
- ahriman.core.formatters.repository_printer module
- ahriman.core.formatters.status_printer module
- ahriman.core.formatters.string_printer module
- ahriman.core.formatters.tree_printer module
- ahriman.core.formatters.update_printer module
- ahriman.core.formatters.user_printer module
- ahriman.core.formatters.validation_printer module
- ahriman.core.formatters.version_printer module
- Module contents
- ahriman.core.gitremote package
- ahriman.core.http package
- ahriman.core.log package
- ahriman.core.report package
- Submodules
- ahriman.core.report.console module
- ahriman.core.report.email module
- ahriman.core.report.html module
- ahriman.core.report.jinja_template module
- ahriman.core.report.remote_call module
- ahriman.core.report.report module
- ahriman.core.report.report_trigger module
- ahriman.core.report.telegram module
- Module contents
- ahriman.core.repository package
- Submodules
- ahriman.core.repository.cleaner module
- ahriman.core.repository.executor module
- ahriman.core.repository.package_info module
- ahriman.core.repository.repository module
- ahriman.core.repository.repository_properties module
RepositoryProperties
RepositoryProperties.configuration
RepositoryProperties.database
RepositoryProperties.ignore_list
RepositoryProperties.pacman
RepositoryProperties.paths
RepositoryProperties.repo
RepositoryProperties.reporter
RepositoryProperties.repository_id
RepositoryProperties.sign
RepositoryProperties.triggers
RepositoryProperties.vcs_allowed_age
RepositoryProperties.packager()
RepositoryProperties.architecture
RepositoryProperties.name
- ahriman.core.repository.update_handler module
- Module contents
- ahriman.core.sign package
- ahriman.core.status package
- Submodules
- ahriman.core.status.client module
Client
Client.load()
Client.package_add()
Client.package_changes_get()
Client.package_changes_set()
Client.package_get()
Client.package_logs()
Client.package_remove()
Client.package_update()
Client.set_building()
Client.set_failed()
Client.set_pending()
Client.set_success()
Client.set_unknown()
Client.status_get()
Client.status_update()
- ahriman.core.status.watcher module
Watcher
Watcher.database
Watcher.repository_id
Watcher.status
Watcher.load()
Watcher.logs_get()
Watcher.logs_remove()
Watcher.logs_update()
Watcher.package_changes_get()
Watcher.package_get()
Watcher.package_remove()
Watcher.package_update()
Watcher.patches_get()
Watcher.patches_remove()
Watcher.patches_update()
Watcher.status_update()
Watcher.packages
- ahriman.core.status.web_client module
- Module contents
- ahriman.core.support package
- ahriman.core.triggers package
- ahriman.core.upload package
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 betterdefault 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 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:
callback (Callable[[argparse.Namespace, str], bool]) – application run function (i.e.
ahriman.application.handlers.handler.Handler.call()
method)args (argparse.Namespace) – command line arguments
repository_id (RepositoryId) – repository unique identifier
process_id (str) – process unique identifier
queue (Queue[ProcessStatus | None]) – output queue
- 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]
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
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
- 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]]
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
isTrue
, second isFalse
- 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. Ifarchitecture
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 toos.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.