ahriman.core.status package

Submodules

ahriman.core.status.client module

class Client

Bases: object

base build status reporter client

static load(repository_id: RepositoryId, configuration: Configuration, database: SQLite | None = None, *, report: bool = True) → Client

load client from settings

Parameters:

  • repository_id (RepositoryId) – repository unique identifier

  • configuration (Configuration) – configuration instance

  • database (SQLite | None, optional) – database instance (Default value = None)

  • report (bool, optional) – force enable or disable reporting (Default value = True)

Returns:

client according to current settings

Return type:

Client

configuration_reload() → None

reload configuration

event_add(event: Event) → None

create new event

Parameters:

event (Event) – audit log event

Raises:

NotImplementedError – not implemented method

event_get(event: str | EventType | None, object_id: str | None, from_date: int | float | None = None, to_date: int | float | None = None, limit: int = -1, offset: int = 0) → list[Event]

retrieve list of events

Parameters:

  • event (str | EventType | None) – filter by event type

  • object_id (str | None) – filter by event object

  • from_date (int | float | None, optional) – minimal creation date, inclusive (Default value = None)

  • to_date (int | float | None, optional) – maximal creation date, exclusive (Default value = None)

  • limit (int, optional) – limit records to the specified count, -1 means unlimited (Default value = -1)

  • offset (int, optional) – records offset (Default value = 0)

Returns:

list of audit log events

Return type:

list[Event]

Raises:

NotImplementedError – not implemented method

logs_rotate(keep_last_records: int) → None

remove older logs from storage

Parameters:

keep_last_records (int) – number of last records to keep

package_changes_get(package_base: str) → Changes

get package changes

Parameters:

package_base (str) – package base to retrieve

Returns:

package changes if available and empty object otherwise

Return type:

Changes

Raises:

NotImplementedError – not implemented method

package_changes_update(package_base: str, changes: Changes) → None

update package changes

Parameters:

  • package_base (str) – package base to update

  • changes (Changes) – changes descriptor

Raises:

NotImplementedError – not implemented method

package_dependencies_get(package_base: str) → Dependencies

get package dependencies

Parameters:

package_base (str) – package base to retrieve

Returns:

package implicit dependencies if available

Return type:

list[Dependencies]

Raises:

NotImplementedError – not implemented method

package_dependencies_update(package_base: str, dependencies: Dependencies) → None

update package dependencies

Parameters:

  • package_base (str) – package base to update

  • dependencies (Dependencies) – dependencies descriptor

Raises:

NotImplementedError – not implemented method

package_get(package_base: str | None) → list[tuple[Package, BuildStatus]]

get package status

Parameters:

package_base (str | None) – package base to get

Returns:

list of current package description and status if it has been found

Return type:

list[tuple[Package, BuildStatus]]

Raises:

NotImplementedError – not implemented method

package_hold_update(package_base: str, *, enabled: bool) → None

update package hold status

Parameters:

  • package_base (str) – package base name

  • enabled (bool) – new hold status

Raises:

NotImplementedError – not implemented method

package_logs_add(log_record: LogRecord) → None

post log record

Parameters:

log_record (LogRecord) – log record

package_logs_get(package_base: str, version: str | None = None, process_id: str | None = None, limit: int = -1, offset: int = 0) → list[LogRecord]

get package logs

Parameters:

  • package_base (str) – package base

  • version (str | None, optional) – package version to search (Default value = None)

  • process_id (str | None, optional) – process identifier to search (Default value = None)

  • limit (int, optional) – limit records to the specified count, -1 means unlimited (Default value = -1)

  • offset (int, optional) – records offset (Default value = 0)

Returns:

package logs

Return type:

list[LogRecord]

Raises:

NotImplementedError – not implemented method

package_logs_remove(package_base: str, version: str | None) → None

remove package logs

Parameters:

  • package_base (str) – package base

  • version (str | None) – package version to remove logs. If None is set, all logs will be removed

Raises:

NotImplementedError – not implemented method

package_patches_get(package_base: str, variable: str | None) → list[PkgbuildPatch]

get package patches

Parameters:

  • package_base (str) – package base to retrieve

  • variable (str | None) – optional filter by patch variable

Returns:

list of patches for the specified package

Return type:

list[PkgbuildPatch]

Raises:

NotImplementedError – not implemented method

package_patches_remove(package_base: str, variable: str | None) → None

remove package patch

Parameters:

  • package_base (str) – package base to update

  • variable (str | None) – patch name. If None is set, all patches will be removed

Raises:

NotImplementedError – not implemented method

package_patches_update(package_base: str, patch: PkgbuildPatch) → None

create or update package patch

Parameters:

  • package_base (str) – package base to update

  • patch (PkgbuildPatch) – package patch

Raises:

NotImplementedError – not implemented method

package_remove(package_base: str) → None

remove packages from watcher

Parameters:

package_base (str) – package base to remove

Raises:

NotImplementedError – not implemented method

package_status_update(package_base: str, status: BuildStatusEnum) → None

update package build status. Unlike package_update() it does not update package properties

Parameters:

  • package_base (str) – package base to update

  • status (BuildStatusEnum) – current package build status

Raises:

NotImplementedError – not implemented method

package_update(package: Package, status: BuildStatusEnum) → None

add new package or update existing one with status

Parameters:
Raises:

NotImplementedError – not implemented method

set_building(package_base: str) → None

set package status to building

Parameters:

package_base (str) – package base to update

set_failed(package_base: str) → None

set package status to failed

Parameters:

package_base (str) – package base to update

set_pending(package_base: str) → None

set package status to pending

Parameters:

package_base (str) – package base to update

set_success(package: Package) → None

set package status to success

Parameters:

package (Package) – current package properties

set_unknown(package: Package) → None

set package status to unknown. Unlike other methods, this method also checks if package is known, and - in case if it is - it silently skips update

Parameters:

package (Package) – current package properties

statistics() → RepositoryStats

get repository statistics

Returns:

repository statistics object

Return type:

RepositoryStats

status_get() → InternalStatus

get internal service status

Returns:

current internal (web) service status

Return type:

InternalStatus

status_update(status: BuildStatusEnum) → None

update ahriman status itself

Parameters:

status (BuildStatusEnum) – current ahriman status

ahriman.core.status.event_bus module

class EventBus(max_size: int)

Bases: LazyLogging

event bus implementation

max_size

maximum size of queue

Type:

int

Parameters:

max_size (int) – maximum size of queue

async broadcast(event_type: EventType, object_id: str | None, **kwargs: Any) → None

broadcast event to all subscribers

Parameters:

  • event_type (EventType) – event type

  • object_id (str | None) – object identifier (e.g. package base)

  • **kwargs (Any) – additional event data

async shutdown() → None

gracefully shutdown all subscribers

async subscribe(topics: list[EventType] | None = None, object_id: str | None = None) → tuple[str, Queue[tuple[str, dict[str, Any]]]]

register new subscriber

Parameters:

  • topics (list[EventType] | None, optional) – list of event types to filter by. If None is set, all events will be delivered (Default value = None)

  • object_id (str | None, optional) – object identifier to filter by. If None is set, events for all objects will be delivered (Default value = None)

Returns:

subscriber identifier and associated queue

Return type:

tuple[str, Queue[SSEvent]]

async unsubscribe(subscriber_id: str) → None

unsubscribe from events

Parameters:

subscriber_id (str) – subscriber unique identifier

ahriman.core.status.local_client module

class LocalClient(repository_id: RepositoryId, database: SQLite)

Bases: Client

local database handler

database

database instance

Type:

SQLite

repository_id

repository unique identifier

Type:

RepositoryId

Parameters:

  • repository_id (RepositoryId) – repository unique identifier

  • database (SQLite) – database instance

event_add(event: Event) → None

create new event

Parameters:

event (Event) – audit log event

event_get(event: str | EventType | None, object_id: str | None, from_date: int | float | None = None, to_date: int | float | None = None, limit: int = -1, offset: int = 0) → list[Event]

retrieve list of events

Parameters:

  • event (str | EventType | None) – filter by event type

  • object_id (str | None) – filter by event object

  • from_date (int | float | None, optional) – minimal creation date, inclusive (Default value = None)

  • to_date (int | float | None, optional) – maximal creation date, exclusive (Default value = None)

  • limit (int, optional) – limit records to the specified count, -1 means unlimited (Default value = -1)

  • offset (int, optional) – records offset (Default value = 0)

Returns:

list of audit log events

Return type:

list[Event]

logs_rotate(keep_last_records: int) → None

remove older logs from storage

Parameters:

keep_last_records (int) – number of last records to keep

package_changes_get(package_base: str) → Changes

get package changes

Parameters:

package_base (str) – package base to retrieve

Returns:

package changes if available and empty object otherwise

Return type:

Changes

package_changes_update(package_base: str, changes: Changes) → None

update package changes

Parameters:

  • package_base (str) – package base to update

  • changes (Changes) – changes descriptor

package_dependencies_get(package_base: str) → Dependencies

get package dependencies

Parameters:

package_base (str) – package base to retrieve

Returns:

package implicit dependencies if available

Return type:

list[Dependencies]

package_dependencies_update(package_base: str, dependencies: Dependencies) → None

update package dependencies

Parameters:

  • package_base (str) – package base to update

  • dependencies (Dependencies) – dependencies descriptor

package_get(package_base: str | None) → list[tuple[Package, BuildStatus]]

get package status

Parameters:

package_base (str | None) – package base to get

Returns:

list of current package description and status if it has been found

Return type:

list[tuple[Package, BuildStatus]]

package_hold_update(package_base: str, *, enabled: bool) → None

update package hold status

Parameters:

  • package_base (str) – package base name

  • enabled (bool) – new hold status

package_logs_add(log_record: LogRecord) → None

post log record

Parameters:

log_record (LogRecord) – log record

package_logs_get(package_base: str, version: str | None = None, process_id: str | None = None, limit: int = -1, offset: int = 0) → list[LogRecord]

get package logs

Parameters:

  • package_base (str) – package base

  • version (str | None, optional) – package version to search (Default value = None)

  • process_id (str | None, optional) – process identifier to search (Default value = None)

  • limit (int, optional) – limit records to the specified count, -1 means unlimited (Default value = -1)

  • offset (int, optional) – records offset (Default value = 0)

Returns:

package logs

Return type:

list[LogRecord]

package_logs_remove(package_base: str, version: str | None) → None

remove package logs

Parameters:

  • package_base (str) – package base

  • version (str | None) – package version to remove logs. If None is set, all logs will be removed

package_patches_get(package_base: str, variable: str | None) → list[PkgbuildPatch]

get package patches

Parameters:

  • package_base (str) – package base to retrieve

  • variable (str | None) – optional filter by patch variable

Returns:

list of patches for the specified package

Return type:

list[PkgbuildPatch]

package_patches_remove(package_base: str, variable: str | None) → None

remove package patch

Parameters:

  • package_base (str) – package base to update

  • variable (str | None) – patch name. If None is set, all patches will be removed

package_patches_update(package_base: str, patch: PkgbuildPatch) → None

create or update package patch

Parameters:

  • package_base (str) – package base to update

  • patch (PkgbuildPatch) – package patch

package_remove(package_base: str) → None

remove packages from watcher

Parameters:

package_base (str) – package base to remove

package_status_update(package_base: str, status: BuildStatusEnum) → None

update package build status. Unlike package_update() it does not update package properties

Parameters:

  • package_base (str) – package base to update

  • status (BuildStatusEnum) – current package build status

Raises:

NotImplementedError – not implemented method

package_update(package: Package, status: BuildStatusEnum) → None

add new package or update existing one with status

Parameters:
Raises:

NotImplementedError – not implemented method

ahriman.core.status.watcher module

class Watcher(client: Client, package_info: PackageInfo, event_bus: EventBus)

Bases: LazyLogging

package status watcher

client

reporter instance

Type:

Client

event_bus

event bus instance

Type:

EventBus

package_info

package info instance

Type:

PackageInfo

status

daemon status

Type:

BuildStatus

Parameters:

  • client (Client) – reporter instance

  • package_info (PackageInfo) – package info instance

  • event_bus (EventBus) – event bus instance

async event_add(event: Event) → None

create new event

Parameters:

event (Event) – audit log event

async event_get(event: str | EventType | None, object_id: str | None, from_date: int | float | None = None, to_date: int | float | None = None, limit: int = -1, offset: int = 0) → list[Event]

retrieve list of events

Parameters:

  • event (str | EventType | None) – filter by event type

  • object_id (str | None) – filter by event object

  • from_date (int | float | None, optional) – minimal creation date, inclusive (Default value = None)

  • to_date (int | float | None, optional) – maximal creation date, exclusive (Default value = None)

  • limit (int, optional) – limit records to the specified count, -1 means unlimited (Default value = -1)

  • offset (int, optional) – records offset (Default value = 0)

Returns:

list of audit log events

Return type:

list[Event]

async load() → None

load packages from local database

async logs_rotate(keep_last_records: int) → None

remove older logs from storage

Parameters:

keep_last_records (int) – number of last records to keep

async package_archives(package_base: str) → list[Package]

get known package archives

Parameters:

package_base (str) – package base

Returns:

list of built package for this package base

Return type:

list[Package]

async package_changes_get(package_base: str) → Changes

get package changes

Parameters:

package_base (str) – package base to retrieve

Returns:

package changes if available and empty object otherwise

Return type:

Changes

async package_changes_update(package_base: str, changes: Changes) → None

update package changes

Parameters:

  • package_base (str) – package base to update

  • changes (Changes) – changes descriptor

async package_dependencies_get(package_base: str) → Dependencies

get package dependencies

Parameters:

package_base (str) – package base to retrieve

Returns:

package implicit dependencies if available

Return type:

list[Dependencies]

async package_dependencies_update(package_base: str, dependencies: Dependencies) → None

update package dependencies

Parameters:

  • package_base (str) – package base to update

  • dependencies (Dependencies) – dependencies descriptor

async package_get(package_base: str) → tuple[Package, BuildStatus]

get current package base build status

Parameters:

package_base (str) – package base

Returns:

package and its status

Return type:

tuple[Package, BuildStatus]

Raises:

UnknownPackageError – if no package found

async package_hold_update(package_base: str, *, enabled: bool) → None

update package hold status

Parameters:

  • package_base (str) – package base name

  • enabled (bool) – new hold status

async package_logs_add(log_record: LogRecord) → None

post log record

Parameters:

log_record (LogRecord) – log record

async package_logs_get(package_base: str, version: str | None = None, process_id: str | None = None, limit: int = -1, offset: int = 0) → list[LogRecord]

get package logs

Parameters:

  • package_base (str) – package base

  • version (str | None, optional) – package version to search (Default value = None)

  • process_id (str | None, optional) – process identifier to search (Default value = None)

  • limit (int, optional) – limit records to the specified count, -1 means unlimited (Default value = -1)

  • offset (int, optional) – records offset (Default value = 0)

Returns:

package logs

Return type:

list[LogRecord]

async package_logs_remove(package_base: str, version: str | None) → None

remove package logs

Parameters:

  • package_base (str) – package base

  • version (str | None) – package version to remove logs. If None is set, all logs will be removed

async package_patches_get(package_base: str, variable: str | None) → list[PkgbuildPatch]

get package patches

Parameters:

  • package_base (str) – package base to retrieve

  • variable (str | None) – optional filter by patch variable

Returns:

list of patches for the specified package

Return type:

list[PkgbuildPatch]

async package_patches_remove(package_base: str, variable: str | None) → None

remove package patch

Parameters:

  • package_base (str) – package base to update

  • variable (str | None) – patch name. If None is set, all patches will be removed

async package_patches_update(package_base: str, patch: PkgbuildPatch) → None

create or update package patch

Parameters:

  • package_base (str) – package base to update

  • patch (PkgbuildPatch) – package patch

async package_remove(package_base: str) → None

remove package base from known list if any

Parameters:

package_base (str) – package base

async package_status_update(package_base: str, status: BuildStatusEnum) → None

update package status

Parameters:

  • package_base (str) – package base to update

  • status (BuildStatusEnum) – new build status

async package_update(package: Package, status: BuildStatusEnum) → None

update package

Parameters:
async packages() → list[tuple[Package, BuildStatus]]

get current known packages list

Returns:

list of packages together with their statuses

Return type:

list[tuple[Package, BuildStatus]]

async shutdown() → None

gracefully shutdown watcher

async status_update(status: BuildStatusEnum) → None

update service status

Parameters:

status (BuildStatusEnum) – new service status

ahriman.core.status.web_client module

class WebClient(repository_id: RepositoryId, configuration: Configuration)

Bases: Client, SyncAhrimanClient

build status reporter web client

repository_id

repository unique identifier

Type:

RepositoryId

Parameters:
static parse_address(configuration: Configuration) → tuple[str, str]

parse address from legacy configuration

Parameters:

configuration (Configuration) – configuration instance

Returns:

tuple of section name and server address

Return type:

tuple[str, str]

configuration_reload() → None

reload configuration

event_add(event: Event) → None

create new event

Parameters:

event (Event) – audit log event

event_get(event: str | EventType | None, object_id: str | None, from_date: int | float | None = None, to_date: int | float | None = None, limit: int = -1, offset: int = 0) → list[Event]

retrieve list of events

Parameters:

  • event (str | EventType | None) – filter by event type

  • object_id (str | None) – filter by event object

  • from_date (int | float | None, optional) – minimal creation date, inclusive (Default value = None)

  • to_date (int | float | None, optional) – maximal creation date, exclusive (Default value = None)

  • limit (int, optional) – limit records to the specified count, -1 means unlimited (Default value = -1)

  • offset (int, optional) – records offset (Default value = 0)

Returns:

list of audit log events

Return type:

list[Event]

logs_rotate(keep_last_records: int) → None

remove older logs from storage

Parameters:

keep_last_records (int) – number of last records to keep

package_changes_get(package_base: str) → Changes

get package changes

Parameters:

package_base (str) – package base to retrieve

Returns:

package changes if available and empty object otherwise

Return type:

Changes

package_changes_update(package_base: str, changes: Changes) → None

update package changes

Parameters:

  • package_base (str) – package base to update

  • changes (Changes) – changes descriptor

package_dependencies_get(package_base: str) → Dependencies

get package dependencies

Parameters:

package_base (str) – package base to retrieve

Returns:

package implicit dependencies if available

Return type:

list[Dependencies]

package_dependencies_update(package_base: str, dependencies: Dependencies) → None

update package dependencies

Parameters:

  • package_base (str) – package base to update

  • dependencies (Dependencies) – dependencies descriptor

package_get(package_base: str | None) → list[tuple[Package, BuildStatus]]

get package status

Parameters:

package_base (str | None) – package base to get

Returns:

list of current package description and status if it has been found

Return type:

list[tuple[Package, BuildStatus]]

package_hold_update(package_base: str, *, enabled: bool) → None

update package hold status

Parameters:

  • package_base (str) – package base name

  • enabled (bool) – new hold status

package_logs_add(log_record: LogRecord) → None

post log record

Parameters:

log_record (LogRecord) – log record

package_logs_get(package_base: str, version: str | None = None, process_id: str | None = None, limit: int = -1, offset: int = 0) → list[LogRecord]

get package logs

Parameters:

  • package_base (str) – package base

  • version (str | None, optional) – package version to search (Default value = None)

  • process_id (str | None, optional) – process identifier to search (Default value = None)

  • limit (int, optional) – limit records to the specified count, -1 means unlimited (Default value = -1)

  • offset (int, optional) – records offset (Default value = 0)

Returns:

package logs

Return type:

list[LogRecord]

package_logs_remove(package_base: str, version: str | None) → None

remove package logs

Parameters:

  • package_base (str) – package base

  • version (str | None) – package version to remove logs. If None is set, all logs will be removed

package_patches_get(package_base: str, variable: str | None) → list[PkgbuildPatch]

get package patches

Parameters:

  • package_base (str) – package base to retrieve

  • variable (str | None) – optional filter by patch variable

Returns:

list of patches for the specified package

Return type:

list[PkgbuildPatch]

package_patches_remove(package_base: str, variable: str | None) → None

remove package patch

Parameters:

  • package_base (str) – package base to update

  • variable (str | None) – patch name. If None is set, all patches will be removed

package_patches_update(package_base: str, patch: PkgbuildPatch) → None

create or update package patch

Parameters:

  • package_base (str) – package base to update

  • patch (PkgbuildPatch) – package patch

package_remove(package_base: str) → None

remove packages from watcher

Parameters:

package_base (str) – basename to remove

package_status_update(package_base: str, status: BuildStatusEnum) → None

update package build status. Unlike package_update() it does not update package properties

Parameters:

  • package_base (str) – package base to update

  • status (BuildStatusEnum) – current package build status

Raises:

NotImplementedError – not implemented method

package_update(package: Package, status: BuildStatusEnum) → None

add new package or update existing one with status

Parameters:
Raises:

NotImplementedError – not implemented method

status_get() → InternalStatus

get internal service status

Returns:

current internal (web) service status

Return type:

InternalStatus

status_update(status: BuildStatusEnum) → None

update ahriman status itself

Parameters:

status (BuildStatusEnum) – current ahriman status

Module contents