85 lines
2.9 KiB
Python
85 lines
2.9 KiB
Python
"""Per event-type severity overrides (admin Settings UI)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.constants.event_types import DEFAULT_EVENT_SEVERITIES
|
|
from app.models.event_severity_override import EventSeverityOverride
|
|
from app.services.notification_settings import VALID_SEVERITIES
|
|
|
|
SOURCE_DB = "db"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EventSeverityOverrideItem:
|
|
event_type: str
|
|
default_severity: str
|
|
override_severity: str | None
|
|
|
|
|
|
def get_override_map(db: Session) -> dict[str, str]:
|
|
rows = db.scalars(select(EventSeverityOverride)).all()
|
|
return {row.event_type: row.severity for row in rows}
|
|
|
|
|
|
def apply_severity_override(payload: dict, db: Session) -> dict:
|
|
"""Return payload copy with severity replaced when override exists."""
|
|
event_type = str(payload.get("type") or "")
|
|
if not event_type:
|
|
return payload
|
|
override_map = get_override_map(db)
|
|
override = override_map.get(event_type)
|
|
if not override:
|
|
return payload
|
|
agent_severity = str(payload.get("severity") or "info")
|
|
if agent_severity == override:
|
|
return payload
|
|
out = dict(payload)
|
|
details = dict(out.get("details") or {})
|
|
details["severity_agent"] = agent_severity
|
|
out["details"] = details
|
|
out["severity"] = override
|
|
return out
|
|
|
|
|
|
def list_severity_override_items(db: Session) -> list[EventSeverityOverrideItem]:
|
|
override_map = get_override_map(db)
|
|
types = set(DEFAULT_EVENT_SEVERITIES) | set(override_map)
|
|
items: list[EventSeverityOverrideItem] = []
|
|
for event_type in sorted(types):
|
|
default = DEFAULT_EVENT_SEVERITIES.get(event_type, "info")
|
|
items.append(
|
|
EventSeverityOverrideItem(
|
|
event_type=event_type,
|
|
default_severity=default,
|
|
override_severity=override_map.get(event_type),
|
|
)
|
|
)
|
|
return items
|
|
|
|
|
|
def replace_severity_overrides(db: Session, overrides: dict[str, str | None]) -> list[EventSeverityOverrideItem]:
|
|
for event_type, severity in overrides.items():
|
|
normalized_type = event_type.strip()
|
|
if not normalized_type:
|
|
raise ValueError("event_type is required")
|
|
if severity is None or severity == "":
|
|
db.execute(
|
|
delete(EventSeverityOverride).where(EventSeverityOverride.event_type == normalized_type)
|
|
)
|
|
continue
|
|
if severity not in VALID_SEVERITIES:
|
|
raise ValueError(f"severity must be one of: {sorted(VALID_SEVERITIES)}")
|
|
row = db.get(EventSeverityOverride, normalized_type)
|
|
if row is None:
|
|
row = EventSeverityOverride(event_type=normalized_type, severity=severity)
|
|
db.add(row)
|
|
else:
|
|
row.severity = severity
|
|
db.flush()
|
|
return list_severity_override_items(db)
|