enpi_api.examples.events

Listen to events

Using the SDK it is possible to listen to events that are emitted by the ENPICOM Platform.

For more information on events, and which events are available, please refer to the Events documentation.

Listen to events

Listen to events that are emitted by the ENPICOM Platform.

This example uses the enpi_api.l2.events.EventListener.loop_forever method to listen indefinitely to events. This function will block the main thread and should only be used in scripts that do not need to do anything besides listening to and acting on events. Alternatively you can use the listener as a context manager, which will listen for events in a background thread.

from enpi_api.l2.events.organization_event_listener import OrganizationEventListener
from enpi_api.l2.types.event import Event


# First we define a callback function that will be called when an event is received
def on_event_callback(topic: str, event: Event) -> None:
    print(f"Received event on topic {topic}: {event}")
    # We could use the event here, we can see for example what type of event was triggered by checking the
    # `category` and `action` fields of the event


# We now define the listener that will listen for events organization wide
listener = OrganizationEventListener(on_event=on_event_callback)

# By calling `loop_forever` the current thread will be blocked, and the listener will keep listening for events, then
# calling the callback function when an event is received
listener.loop_forever()

View Source

 1'''
 2# Listen to events
 3
 4Using the SDK it is possible to listen to events that are emitted by the ENPICOM Platform.
 5
 6For more information on events, and which events are available, please refer to the [Events](/public-api/v1/public/docs/events/) documentation.
 7
 8
 9##Listen to events
10
11Listen to events that are emitted by the ENPICOM Platform.
12
13This example uses the `enpi_api.l2.events.EventListener.loop_forever` method to listen indefinitely to events. This function will block
14the main thread and should only be used in scripts that do not need to do anything besides listening to and acting on events. Alternatively
15you can use the listener as a context manager, which will listen for events in a background thread.
16
17```python
18from enpi_api.l2.events.organization_event_listener import OrganizationEventListener
19from enpi_api.l2.types.event import Event
20
21
22# First we define a callback function that will be called when an event is received
23def on_event_callback(topic: str, event: Event) -> None:
24    print(f"Received event on topic {topic}: {event}")
25    # We could use the event here, we can see for example what type of event was triggered by checking the
26    # `category` and `action` fields of the event
27
28
29# We now define the listener that will listen for events organization wide
30listener = OrganizationEventListener(on_event=on_event_callback)
31
32# By calling `loop_forever` the current thread will be blocked, and the listener will keep listening for events, then
33# calling the callback function when an event is received
34listener.loop_forever()
35
36```
37'''