Source code for workflows.events

"""Event definitions for the workflows app.

This module centralizes the set of supported workflow events and provides
helpers to enumerate protocols and operations.
"""

from dataclasses import dataclass


@dataclass(frozen=True)
[docs] class Event: """Represents a single event. Attributes: key: Internal name for the event. protocol: Protocol name (e.g., "EST", "CMP", "SCEP"); may be empty. operation: Protocol operation (e.g., "simpleenroll"); may be empty. handler: Service key that will process the event. """
[docs] key: str
[docs] protocol: str
[docs] operation: str
[docs] handler: str
[docs] class Events: """Central definition of all supported events in one place."""
[docs] est_simpleenroll = Event( key='est_simpleenroll', protocol='est', operation='simpleenroll', handler='certificate_request', )
[docs] est_simplereenroll = Event( key='est_simplereenroll', protocol='EST', operation='simplereenroll', handler='certificate_request', )
""" est_cacerts = Event( key='est_cacerts', protocol='EST', operation='', handler='certificate_issued', ) est_csrattrs = Event( key='est_csrattrs', protocol='EST', operation='csrattrs', handler='certificate_request', ) cmp_certrequest = Event( key='cmp_certrequest', protocol='CMP', operation='certRequest', handler='certificate_request', ) cmp_revocationrequest = Event( key='cmp_revocationrequest', protocol='CMP', operation='revocationRequest', handler='certificate_request', ) scep_pkioperation = Event( key='scep_pkioperation', protocol='SCEP', operation='PKIOperation', handler='certificate_request', )"""
[docs] device_created = Event( key='device_created', protocol='device', operation='created', handler='device_action', )
[docs] device_onboarded = Event( key='device_onboarded', protocol='device', operation='onboard', handler='device_action', )
[docs] device_deleted = Event( key='device_deleted', protocol='device', operation='deleted', handler='device_action', )
@classmethod
[docs] def all(cls) -> list[Event]: """Return every Event defined on this class. Returns: list[Event]: All Event instances defined as class attributes. """ return [v for v in vars(cls).values() if isinstance(v, Event)]
@classmethod
[docs] def protocols(cls) -> list[str]: """Return the unique list of non-empty protocol names. Returns: list[str]: Sorted list of protocol identifiers used by defined events. """ return sorted({e.protocol for e in cls.all() if e.protocol})
@classmethod
[docs] def operations_for(cls, protocol: str) -> list[str]: """Return all operations available for a given protocol. Args: protocol: Protocol identifier (e.g. ``"est"``). Returns: list[str]: Operations associated with the given protocol. """ return [e.operation for e in cls.all() if e.protocol == protocol]