majuna/app/alarms.py

import datetime
from typing import Optional

from app.extensions import db
from app.models.alarms import Alarm


def _get_alarm(target: str,
               alarm_type: str,
               *,
               proxy_id: Optional[int] = None,
               origin_id: Optional[int] = None,
               create_if_missing: bool = True) -> Optional[Alarm]:
    alarm: Optional[Alarm]
    if target == "proxy":
        alarm = Alarm.query.filter(
            Alarm.target == "proxy",
            Alarm.alarm_type == alarm_type,
            Alarm.proxy_id == proxy_id
        ).first()
    elif target == "origin":
        alarm = Alarm.query.filter(
            Alarm.target == "origin",
            Alarm.alarm_type == alarm_type,
            Alarm.proxy_id == origin_id
        ).first()
    else:
        return None
    if create_if_missing and alarm is None:
        alarm = Alarm()
        alarm.target = target
        alarm.alarm_type = alarm_type
        alarm.state_changed = datetime.datetime.utcnow()
        if target == "proxy":
            alarm.proxy_id = proxy_id
        if target == "origin":
            alarm.origin_id = origin_id
        db.session.add(alarm)
        db.session.commit()
    return alarm


def get_proxy_alarm(proxy_id: int, alarm_type: str) -> Alarm:
    alarm = _get_alarm("proxy", alarm_type, proxy_id=proxy_id)
    assert(alarm is not None)
    return alarm