pyfebiopt.monitoring.registry

Manage the JSONL-backed registry of optimization runs.

Exceptions

ActiveRunDeletionError

Raised when attempting to delete an optimization run that is still active.

Classes

RunRegistry

Thread-safe registry storing optimization run snapshots backed by JSONL.

Module Contents

exception pyfebiopt.monitoring.registry.ActiveRunDeletionError(run_id: str)

Bases: RuntimeError

Raised when attempting to delete an optimization run that is still active.

Describe the active run that blocked deletion.

Parameters:

run_id – Identifier of the active run.

run_id

class pyfebiopt.monitoring.registry.RunRegistry(db_path: pathlib.Path, *, max_history: int | None = None)

Thread-safe registry storing optimization run snapshots backed by JSONL.

Initialize the registry with the backing database path.

db_path

list_runs() → list[pyfebiopt.monitoring.runstate.OptimizationRun]

Return all tracked optimization runs.

get_run(run_id: str) → pyfebiopt.monitoring.runstate.OptimizationRun | None

Return the run snapshot for the given identifier.

apply_event(run_id: str, event: str, payload: dict[str, object], ts: float) → None

Incorporate an incoming event into the relevant run.

refresh() → None

Reload the registry contents from disk.

delete_run(run_id: str, *, force: bool = False) → bool

Remove a run if it is safe to delete or forcefully requested.

Returns:

True if deleted, False otherwise.

Return type:

bool

clear(*, force: bool = False) → list[str]

Drop runs and list those kept either due to activity or force=False.

Returns:

Run ids that were not removed.

Return type:

list[str]