patrickandClaude Sonnet 5.5
347f55cf7c
Includes atomic migrations, per-rule reminder semantics, first-run helper script and the note on the undecided UI chat idea. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
39 lines
1.4 KiB
Python
39 lines
1.4 KiB
Python
from typing import Mapping, Sequence
|
|
|
|
from ..checkmk import ServiceState
|
|
from ..db import Database
|
|
from ..models import STATE_OK, Key, StateRow
|
|
from .events import build_row
|
|
|
|
BASELINE_DONE_KEY = "baseline.done"
|
|
BASELINE_REQUEST_KEY = "baseline.requested"
|
|
FLAG = "1"
|
|
|
|
|
|
def baseline_rows(previous: Mapping[Key, StateRow], current: Sequence[ServiceState], now: int) -> list[StateRow]:
|
|
return [build_row(cur, previous.get((cur.host, cur.service)), now, cur.state, cur.flapping,
|
|
(now, cur.last_state_change))
|
|
for cur in current if cur.hard]
|
|
|
|
|
|
def open_rows(rows: Sequence[StateRow]) -> list[StateRow]:
|
|
return [row for row in rows if row.state != STATE_OK]
|
|
|
|
|
|
class BaselineGuard:
|
|
def __init__(self, db: Database):
|
|
self._db = db
|
|
|
|
def _is_set(self, key: str) -> bool:
|
|
return bool(self._db.query("SELECT 1 FROM app_state WHERE key=?", (key,)))
|
|
|
|
def needed(self, state_empty: bool) -> bool:
|
|
return self._is_set(BASELINE_REQUEST_KEY) or (state_empty and not self._is_set(BASELINE_DONE_KEY))
|
|
|
|
def request(self) -> None:
|
|
self._db.execute("INSERT OR REPLACE INTO app_state(key, value) VALUES (?, ?)", (BASELINE_REQUEST_KEY, FLAG))
|
|
|
|
def completion(self) -> list[tuple[str, tuple]]:
|
|
return [("INSERT OR REPLACE INTO app_state(key, value) VALUES (?, ?)", (BASELINE_DONE_KEY, FLAG)),
|
|
("DELETE FROM app_state WHERE key=?", (BASELINE_REQUEST_KEY,))]
|