Example implementation of a wrapper for Observium alerts

.py

from __future__ import annotations

import datetime
from collections import OrderedDict
from dataclasses import dataclass, field
from types import SimpleNamespace

StableAlertID = str


@dataclass
class Alert(SimpleNamespace):
    data: OrderedDict[str, str | datetime.datetime] = field(default_factory=OrderedDict)
    _OBSERVIUM_PREFIX: str = 'OBSERVIUM_'

    ...

    @classmethod
    def from_env(cls, runtime: SimpleNamespace) -> Alert:
        payload: OrderedDict[str, str | datetime.datetime] = OrderedDict(
            {
                k: v
                for k, v in os.environ.items()
                if k.startswith(Alert._OBSERVIUM_PREFIX)
            }
        )
        
        # ... convert values like timestamps to datetime
        
        # ... verify mandatory fields are present

        return cls(payload, ...)

    @property
    def stable_alert_id(self) -> StableAlertID:
        '''
        Returns the most "Stable Alert ID" available stable across
        alert/reminder/recovery for the same device/alert name/entity.

        WARNING: OBSERVIUM_ALERT_ID is NOT stable, the entire alert
                 association database is rebuilt whenever we change any
                 alert associations.
        '''

        # NOTE: Any change to the keys used for stable ID generation should be
        #       considered a breaking change, since it would cause all
        #       existing ticket associations to break.
        # Maybe better way, wanted keys in order of preference with fallbacks
        wanted_keys = [
            (
                'ALERT_TEST_ID',  # - new field added 2026-03-23 by Adama at my
                #   request, which is the ID of the alert test that
                #   triggered this alert)
                'ALERT_NAME',
            ),
            ('DEVICE_TYPE',),  # - e.g. unknown, firewall, network, server,
            #        wireless
            ('DEVICE_ID',),  # - unique id for each device
            ('ENTITY_TYPE',),  # - e.g. processor, status, port, device
            ('ENTITY_ID',),  # - unique id for each entity
        ]

        seperator = '='

        res = []
        for key_position, wanted_key in enumerate(wanted_keys, start=1):
            found = False
            for key_option in wanted_key:
                if key_option in self.data:
                    value = self.data[key_option]
                    res.append(value)
                    found = True
                    break
                else:
                    self.log.warning(
                        f'Alert stableAlertID key {key_option=} not found for stable ID generation at position {key_position}'
                    )
            if not found:
                self.log.error(
                    f'Alert stableAlertID has missing keys {wanted_key} for stable ID generation at position {key_position}, using placeholder value'
                )
                res.append(f'<{wanted_key[0]}=MISSING>')

        stable_id_result = seperator.join(res)

        assert stable_id_result, '"Stable Alert ID" must never be empty'
        return stable_id_result