akkudoktoreos.server.retentionmanager.JobState

class akkudoktoreos.server.retentionmanager.JobState(name: str, func: Callable[[], None] | Callable[[], Coroutine[Any, Any, None]], interval_attr: str, fallback_interval: float, config_getter: Callable[[str], Any], on_exception: Callable[[Exception], None] | Callable[[Exception], Coroutine[Any, Any, None]] | None = None, last_run_at: float = 0.0, last_duration: float = 0.0, last_error: str | None = None, run_count: int = 0, is_running: bool = False)

Bases: object

Runtime state tracked for a single managed job.

name

Unique human-readable job name used in logs and metrics.

Type:

str

func

The maintenance callable. Must accept no arguments.

Type:

Callable[[], None] | Callable[[], Coroutine[Any, Any, None]]

interval_attr

Key passed to config_getter to retrieve the interval in seconds for this job.

Type:

str

fallback_interval

Interval in seconds used when the key is not found or returns zero.

Type:

float

config_getter

Callable that accepts a string key and returns the corresponding configuration value. Invoked with interval_attr to obtain the interval in seconds.

Type:

Callable[[str], Any]

on_exception

Optional callable invoked with the raised exception whenever func fails. May be sync or async.

Type:

Callable[[Exception], None] | Callable[[Exception], Coroutine[Any, Any, None]] | None

last_run_at

Monotonic timestamp of the last completed run; 0.0 means never run.

Type:

float

last_duration

How long the last run took, in seconds.

Type:

float

last_error

String representation of the last exception, or None if the last run succeeded.

Type:

str | None

run_count

Total number of completed runs (successful or not).

Type:

int

is_running

True while the job coroutine is currently executing.

Type:

bool

__init__(name: str, func: Callable[[], None] | Callable[[], Coroutine[Any, Any, None]], interval_attr: str, fallback_interval: float, config_getter: Callable[[str], Any], on_exception: Callable[[Exception], None] | Callable[[Exception], Coroutine[Any, Any, None]] | None = None, last_run_at: float = 0.0, last_duration: float = 0.0, last_error: str | None = None, run_count: int = 0, is_running: bool = False) → None

Methods

__init__(name, func, interval_attr, ...[, ...])
interval()	Retrieve the current interval by calling config_getter with interval_attr.
is_due()	Check whether enough time has elapsed since the last run to execute this job again.
summary()	Build a serialisable snapshot of the job's current state.

Attributes

is_running
last_duration
last_error
last_run_at
on_exception
run_count
name
func
interval_attr
fallback_interval
config_getter

name: str

func: Callable[[], None] | Callable[[], Coroutine[Any, Any, None]]

interval_attr: str

fallback_interval: float

config_getter: Callable[[str], Any]

on_exception: Callable[[Exception], None] | Callable[[Exception], Coroutine[Any, Any, None]] | None = None

last_run_at: float = 0.0

last_duration: float = 0.0

last_error: str | None = None

run_count: int = 0

is_running: bool = False

interval() → float | None

Retrieve the current interval by calling config_getter with interval_attr.

Returns None when the config value is None, which signals that the job is disabled and must never fire. Falls back to fallback_interval when the key is not found.

Returns:

The interval in seconds, or None if the job is disabled.

is_due() → bool

Check whether enough time has elapsed since the last run to execute this job again.

Returns False immediately when interval returns None (job is disabled), so a disabled job never fires regardless of when it last ran.

Returns:

True if the job should be executed on this tick, False otherwise.

summary() → dict

Build a serialisable snapshot of the job’s current state.

Returns:

A dictionary suitable for JSON serialisation, containing the job name, interval key, last run timestamp, last duration, last error, run count, and whether the job is currently running.

__init__(name: str, func: Callable[[], None] | Callable[[], Coroutine[Any, Any, None]], interval_attr: str, fallback_interval: float, config_getter: Callable[[str], Any], on_exception: Callable[[Exception], None] | Callable[[Exception], Coroutine[Any, Any, None]] | None = None, last_run_at: float = 0.0, last_duration: float = 0.0, last_error: str | None = None, run_count: int = 0, is_running: bool = False) → None