Skip to content

Event and FSM

Signed domain events and an event-triggered finite state machine. Narrative guide: Event and FSM.

Event

pypepper.event.event

Signed domain events with JSON marshal.

Header

Bases: IHeader

Source code in pypepper/event/event.py 
15
16
17
18
19
20
21
22
23
24
class Header(IHeader):
    default_version = "1"

    def __init__(self):
        self.namespace = ""
        self.request_id = ""
        self.sender = ""
        self.id = uuid.new_uuid()
        self.timestamp = time.get_utc_datetime()
        self.version = self.default_version

default_version class-attribute instance-attribute 

default_version = '1'

namespace instance-attribute 

namespace = ''

request_id instance-attribute 

request_id = ''

sender instance-attribute 

sender = ''

id instance-attribute 

id = uuid.new_uuid()

timestamp instance-attribute 

timestamp = time.get_utc_datetime()

version instance-attribute 

version = self.default_version

Payload

Bases: IPayload

Source code in pypepper/event/event.py 
27
28
29
30
31
32
33
34
35
class Payload(IPayload):
    def __init__(self, dict_: dict | None = None):
        self.id = ""
        self.category = ""
        self.digest = None
        self.raw = None
        if dict_:
            for key in dict_:
                setattr(self, key, dict_[key])

id instance-attribute 

id = ''

category instance-attribute 

category = ''

digest instance-attribute 

digest = None

raw instance-attribute 

raw = None

Data

Bases: IData

Source code in pypepper/event/event.py 
38
39
40
41
42
43
44
class Data(IData):
    def __init__(self):
        self.flow = ""
        self.name = ""
        self.src = ""
        self.header = Header()
        self.payload = Payload()

flow instance-attribute 

flow = ''

name instance-attribute 

name = ''

src instance-attribute 

src = ''

header instance-attribute 

header = Header()

payload instance-attribute 

payload = Payload()

Event

Bases: IEvent

Source code in pypepper/event/event.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
class Event(IEvent):
    def __init__(self, data: IData | None = None, sig: bytes | None = None):
        if not data:
            self.data = Data()
        else:
            self.data = data
        self.signature = sig

    def set_event_id(self, event_id: str):
        self.data.header.id = event_id

    def set_event_namespace(self, namespace: str):
        self.data.header.namespace = namespace

    def set_event_version(self, version: str):
        self.data.header.version = version

    def set_request_id(self, req_id: str):
        self.data.header.request_id = req_id

    def set_sender(self, sender: str):
        self.data.header.sender = sender

    def set_flow(self, flow: str):
        self.data.flow = flow

    def set_name(self, name: str):
        self.data.name = name

    def set_src(self, src: str):
        self.data.src = src

    def set_payload(self, payload: IPayload):
        self.data.payload = payload

    def add_payload(self, payload_id: str, category: str, raw: bytes, hash_alg: str | None = None) -> None:
        assert payload_id, "payload ID is empty"
        assert category, "category is empty"
        assert raw, "payload raw is empty"

        self.data.payload.id = payload_id
        self.data.payload.category = category
        self.data.payload.raw = raw

        if hash_alg:
            self.data.payload.digest = digest.get(raw, hash_alg)

    def _canonical_bytes(self) -> bytes:
        """Stable JSON encoding used for signing and marshaling."""
        payload = _data_to_dict(self.data)
        return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")

    def sign(self, certificate: bytes, hash_alg: str, passphrase: bytes | None = None) -> bytes:
        sig = ecdsa.sign(self._canonical_bytes(), certificate, hash_alg, passphrase)
        self.signature = sig
        return sig

    def verify(self, certificate: bytes, hash_alg: str) -> bool:
        if self.signature is None:
            return False
        return ecdsa.verify(self._canonical_bytes(), certificate, self.signature, hash_alg)

    def marshal(self) -> str:
        body = _data_to_dict(self.data)
        envelope = {
            "data": body,
            "signature": base64.b64encode(self.signature).decode("ascii") if self.signature else None,
        }
        return json.dumps(envelope, sort_keys=True, separators=(",", ":"), ensure_ascii=False)

data instance-attribute 

data = Data()

signature instance-attribute 

signature = sig

set_event_id 

set_event_id(event_id: str)
Source code in pypepper/event/event.py 
85
86
def set_event_id(self, event_id: str):
    self.data.header.id = event_id

set_event_namespace 

set_event_namespace(namespace: str)
Source code in pypepper/event/event.py 
88
89
def set_event_namespace(self, namespace: str):
    self.data.header.namespace = namespace

set_event_version 

set_event_version(version: str)
Source code in pypepper/event/event.py 
91
92
def set_event_version(self, version: str):
    self.data.header.version = version

set_request_id 

set_request_id(req_id: str)
Source code in pypepper/event/event.py 
94
95
def set_request_id(self, req_id: str):
    self.data.header.request_id = req_id

set_sender 

set_sender(sender: str)
Source code in pypepper/event/event.py 
97
98
def set_sender(self, sender: str):
    self.data.header.sender = sender

set_flow 

set_flow(flow: str)
Source code in pypepper/event/event.py 
100
101
def set_flow(self, flow: str):
    self.data.flow = flow

set_name 

set_name(name: str)
Source code in pypepper/event/event.py 
103
104
def set_name(self, name: str):
    self.data.name = name

set_src 

set_src(src: str)
Source code in pypepper/event/event.py 
106
107
def set_src(self, src: str):
    self.data.src = src

set_payload 

set_payload(payload: IPayload)
Source code in pypepper/event/event.py 
109
110
def set_payload(self, payload: IPayload):
    self.data.payload = payload

add_payload 

add_payload(
    payload_id: str,
    category: str,
    raw: bytes,
    hash_alg: str | None = None,
) -> None
Source code in pypepper/event/event.py 
112
113
114
115
116
117
118
119
120
121
122
def add_payload(self, payload_id: str, category: str, raw: bytes, hash_alg: str | None = None) -> None:
    assert payload_id, "payload ID is empty"
    assert category, "category is empty"
    assert raw, "payload raw is empty"

    self.data.payload.id = payload_id
    self.data.payload.category = category
    self.data.payload.raw = raw

    if hash_alg:
        self.data.payload.digest = digest.get(raw, hash_alg)

sign 

sign(
    certificate: bytes,
    hash_alg: str,
    passphrase: bytes | None = None,
) -> bytes
Source code in pypepper/event/event.py 
129
130
131
132
def sign(self, certificate: bytes, hash_alg: str, passphrase: bytes | None = None) -> bytes:
    sig = ecdsa.sign(self._canonical_bytes(), certificate, hash_alg, passphrase)
    self.signature = sig
    return sig

verify 

verify(certificate: bytes, hash_alg: str) -> bool
Source code in pypepper/event/event.py 
134
135
136
137
def verify(self, certificate: bytes, hash_alg: str) -> bool:
    if self.signature is None:
        return False
    return ecdsa.verify(self._canonical_bytes(), certificate, self.signature, hash_alg)

marshal 

marshal() -> str
Source code in pypepper/event/event.py 
139
140
141
142
143
144
145
def marshal(self) -> str:
    body = _data_to_dict(self.data)
    envelope = {
        "data": body,
        "signature": base64.b64encode(self.signature).decode("ascii") if self.signature else None,
    }
    return json.dumps(envelope, sort_keys=True, separators=(",", ":"), ensure_ascii=False)

new 

new(
    name: str | None = None, src: str | None = None
) -> Event
Source code in pypepper/event/event.py 
148
149
150
151
152
153
154
def new(name: str | None = None, src: str | None = None) -> Event:
    evt = Event()
    if name is not None:
        evt.set_name(name)
    if src is not None:
        evt.set_src(src)
    return evt

FSM

pypepper.fsm.fsm

Event-triggered finite state machine with rollback.

State

Bases: IState

Machine State

Source code in pypepper/fsm/fsm.py 
15
16
17
18
19
20
21
class State(IState):
    """
    Machine State
    """

    def __init__(self, value: str):
        self.value = value

value instance-attribute 

value = value

Transition

Bases: ITransition

State Transition

Source code in pypepper/fsm/fsm.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class Transition(ITransition):
    """
    State Transition
    """

    def __init__(
        self,
        event: IEvent,
        from_state: Collection[IState],
        to_state: IState,
        handler: Callable[..., Any] | None = None,
        context: MutableMapping[Any, Any] | None = None,
    ):
        self.event = event
        self.from_state = from_state
        self.to_state = to_state
        self.handler = handler
        self.context = context

event instance-attribute 

event = event

from_state instance-attribute 

from_state = from_state

to_state instance-attribute 

to_state = to_state

handler instance-attribute 

handler = handler

context instance-attribute 

context = context

Options

Bases: IOptions

FSM Options

Source code in pypepper/fsm/fsm.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
class Options(IOptions):
    """
    FSM Options
    """

    def __init__(
        self,
        fsm_id: str,
        initial: IState,
        transitions: Collection[ITransition],
    ):
        self.fsm_id = fsm_id
        self.initial = initial
        self.transitions = transitions

fsm_id instance-attribute 

fsm_id = fsm_id

initial instance-attribute 

initial = initial

transitions instance-attribute 

transitions = transitions

Target

Bases: ITarget

Target state with handler

Source code in pypepper/fsm/fsm.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
class Target(ITarget):
    """
    Target state with handler
    """

    def __init__(
        self,
        state: IState,
        handler: Callable[..., Any] | None = None,
        context: MutableMapping[Any, Any] | None = None,
    ):
        self.state = state
        self.handler = handler
        self.context = context

state instance-attribute 

state = state

handler instance-attribute 

handler = handler

context instance-attribute 

context = context

Response

Bases: IResponse

Transition response

Source code in pypepper/fsm/fsm.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
class Response(IResponse):
    """
    Transition response
    """

    def __init__(
        self,
        state: IState,
        error: Any,
        event_handler_result: Any | None = None,
        transition_result: Any | None = None,
    ):
        self.state = state
        self.error = error
        self.event_handler_result = event_handler_result
        self.transition_result = transition_result

state instance-attribute 

state = state

error instance-attribute 

error = error

event_handler_result instance-attribute 

event_handler_result = event_handler_result

transition_result instance-attribute 

transition_result = transition_result

FSM

Bases: IFSM

Finite State Machine

Source code in pypepper/fsm/fsm.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
class FSM(IFSM):
    """
    Finite State Machine
    """

    def __init__(self, options: IOptions):
        self._id = options.fsm_id
        self._current: IState | None = options.initial
        self._transitions: MutableMapping[str, ITarget] = {}
        self._events: MutableMapping[str, IEvent] = {}
        self._states: MutableMapping[IState, bool] = {}

        for tr in options.transitions:
            for from_state in tr.from_state:
                self._transitions[self._build_transition_key(tr.event, from_state)] = Target(
                    state=tr.to_state,
                    handler=tr.handler,
                    context=tr.context,
                )
                self._states[from_state] = True
            self._states[tr.to_state] = True
            self._events[self._build_event_key(tr.event)] = tr.event

    @staticmethod
    def _build_transition_key(event: IEvent, from_state: IState) -> str:
        """
        Build transition key
        :param event: event
        :param from_state: from some state (source state)
        :return: the transition key in JSON style
        """

        return json.dumps(
            {
                "flow": event.data.flow,
                "name": event.data.name,
                "from_state": from_state.value,
            }
        )

    @staticmethod
    def _build_event_key(event: IEvent) -> str:
        """
        Build event key
        :param event: event
        :return: the event key in JSON style
        """

        return json.dumps(
            {
                "flow": event.data.flow,
                "name": event.data.name,
            }
        )

    def _transition(
        self,
        event: IEvent,
        handler: Callable[..., Any] | None = None,
        context: MutableMapping[Any, Any] | None = None,
    ) -> Response:
        """
        Transition state
        :param event: event
        :param handler: transition handler
        :param context: transition handler's context
        :return: transition response
        """

        if self._current is None:
            raise InternalException(ERROR_INVALID_EVENT)

        key = self._build_transition_key(event, self._current)

        target = self._transitions.get(key)
        if not target:
            return Response(
                state=self._current,
                error=InternalException(ERROR_INVALID_EVENT),
            )

        prev = self._current
        self._current = target.state

        event_handler_result: Any | None = None
        if target.handler:
            try:
                event_handler_result = target.handler(target.context) if target.context else target.handler()
            except Exception as e:
                self._current = prev
                return Response(
                    state=self._current,
                    error=e,
                )

        transition_result: Any | None = None
        if handler:
            try:
                transition_result = handler(context) if context else handler()
            except Exception as e:
                self._current = prev
                return Response(
                    state=self._current,
                    error=e,
                    event_handler_result=event_handler_result,
                )

        return Response(
            state=self._current,
            error=None,
            event_handler_result=event_handler_result,
            transition_result=transition_result,
        )

    def current(self) -> IState | None:
        """
        Get FSM current state.
        :return: current state (None after close)
        """

        return self._current

    def restore(self, state: IState | None) -> None:
        """Restore current state without running a transition (rollback helper)."""
        self._current = state

    def on(
        self,
        event: IEvent,
        handler: Callable[..., Any] | None = None,
        context: MutableMapping[Any, Any] | None = None,
    ) -> IResponse:
        """
        Trigger event
        :param event: event
        :param handler: transition handler
        :param context: transition handler's context
        :return: transition response
        """

        return self._transition(event, handler, context)

    def close(self):
        """
        Close FSM (unsafe)
        :return: None
        """

        self._transitions.clear()
        self._events.clear()
        self._states.clear()
        self._current = None

current 

current() -> IState | None

Get FSM current state. :return: current state (None after close)

Source code in pypepper/fsm/fsm.py 
208
209
210
211
212
213
214
def current(self) -> IState | None:
    """
    Get FSM current state.
    :return: current state (None after close)
    """

    return self._current

restore 

restore(state: IState | None) -> None

Restore current state without running a transition (rollback helper).

Source code in pypepper/fsm/fsm.py 
216
217
218
def restore(self, state: IState | None) -> None:
    """Restore current state without running a transition (rollback helper)."""
    self._current = state

on 

on(
    event: IEvent,
    handler: Callable[..., Any] | None = None,
    context: MutableMapping[Any, Any] | None = None,
) -> IResponse

Trigger event

Parameters:

Name Type Description Default
event IEvent

event

 required
handler Callable[..., Any] | None

transition handler

 None
context MutableMapping[Any, Any] | None

transition handler's context

 None

Returns:

Type Description
IResponse

transition response
Source code in pypepper/fsm/fsm.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def on(
    self,
    event: IEvent,
    handler: Callable[..., Any] | None = None,
    context: MutableMapping[Any, Any] | None = None,
) -> IResponse:
    """
    Trigger event
    :param event: event
    :param handler: transition handler
    :param context: transition handler's context
    :return: transition response
    """

    return self._transition(event, handler, context)

close 

close()

Close FSM (unsafe) :return: None

Source code in pypepper/fsm/fsm.py 
236
237
238
239
240
241
242
243
244
245
def close(self):
    """
    Close FSM (unsafe)
    :return: None
    """

    self._transitions.clear()
    self._events.clear()
    self._states.clear()
    self._current = None

new 

new(options: IOptions) -> FSM

New an FSM.

Parameters:

Name Type Description Default
options IOptions

FSM options.

 required

Returns:

Type Description
FSM

instance of FSM.
Source code in pypepper/fsm/fsm.py 
248
249
250
251
252
253
254
255
def new(options: IOptions) -> FSM:
    """
    New an FSM.
    :param options: FSM options.
    :return: instance of FSM.
    """

    return FSM(options)