Events
The event system allows you to listen to actions that happen inside of the ENPICOM Platform through MQTT over Websockets. Each event is a JSON object that contains information about the event that occurred.
Connecting
To be able to connect to the event system, you will need to have an API key.
The easiest way to connect to the event system is to use our Python SDK. Alternatively, any library that can connect to MQTT over Websockets can be used to connect to the event system, an example in Python is the paho-mqtt library.
If you choose to implement the connection yourself, you will need to connect to the following URL: igx.bio/events and send the API key in
the enpi-api-key header.
Click the button below to show example code on how to connect using our ENPICOM Python SDK.
Show me the code!
from enpi_api.l2.events import OrganizationEventListener
from enpi_api.l2.types.event import Event
# Define a callback function that is called whenever an event is received. You will receive the exact topic it was
# received on, and the event itself.
def on_event_callback(topic: str, event: Event) -> None:
print(f"Received event on topic {topic}: {event.model_dump_json(indent=2)}")
# You can listen to either organization events, or your user events if you have an API key linked to a user.
# The `...EventListener` class can be used in two ways, both automatically reconnecting on connection failures.
# The first method is to create an instance and call `loop_forever` on it, which will block the current thread. Which is
# shown below, this is useful for scripts that do not need to do anything else other than listening for events.
listener = OrganizationEventListener(on_event=on_event_callback)
listener.loop_forever()
# Alternatively, you can use the `...EventListener` as a context manager, which will listen in the background, and
# automatically stop listening when the context is exited. Inside the context block you need to loop or wait yourself
# to keep the context alive.
with OrganizationEventListener(on_event=on_event_callback) as listener:
# Keep the context alive, for example by sleeping, or running a loop.
import time
should_run = True
while should_run:
time.sleep(1)
print("Still alive")
Event Structure
When you receive an event, you receive a JSON structure (or a Python object, if you are using the ENPICOM Python SDK) with the following properties:
| Property | Type | Description |
|---|---|---|
| timestamp | string/datetime | A timestamp indicating when the event was fired |
| organization_id | integer | The unique identifier of the organization |
| user_id | integer (optional) | The unique identifier of the user associated ot this event, if any |
| category | string | The category to which the event belongs to, see the list of events below |
| action | string | The action that was performed, see the list of events below |
| payload | json/dict | A set of key/value pairs containing information about the event |
List of events
Below is a list of all events, including a description of their payload.
The headers are formatted as {category}.{action}.
collection.created
Occurs when a new collection has been created.
payload
| collection_id | integer | The unique identifier of the collection that was created |
collection.updated
Occurs when an existing collection has been updated. This also occurs when a clone or sequence inside a collection has been updated.
payload
| collection_id | integer | The unique identifier of the collection that was updated |
collection.deleted
Occurs when a collection has been deleted.
payload
| collection_id | integer | The unique identifier of the collection that was deleted |