Custom events¶
It is possible to extend the events subscribable to in Lime Webhooks by registering custom events via a Python entry point.
Types of events¶
custom_event- A totally custom event with any routing key and payloadlimeobject_event- An extension to the built in limeobject events
Registering events¶
In a Python package named custom-events.
Expose a variable like:
my_events = {
'custom_events': [
{
'name': 'my-event.test',
'namespace': ['my', 'event'],
'version': 'v1',
'scope': 'Custom Event'
}
],
'limeobject_events': [
{
'name': 'customer',
'limetype': 'company',
'events': ['new', 'update']
}
]
}
In this case we add it into our package __init__.py to you can put it in any file that makes sense. The different nodes custom_events and limeobject_events are optional if you are not registering events of that type.
Set up the entry point in your poetry based pyproject.toml as (skip the comment).
[tool.poetry.plugins."lime.webhooks.events"]
# unique name path variable
# | | |
"custom-events" = 'custom_event:my_events'
Custom event¶
A custom event is an event that can be published on any routing key and with any payload. The definition is:
name- Your events unique name. Most be globally uniquenamespace- your package event namespaceversion- version of your eventscope- A label for your module that will show up in Lime Admin
Given the example above a event should be published on the routing key my.event.my-event.test.v1
Limeobject event¶
A very common scenario is to publish an event that extends our built in lime object events. This for example if we would like an event to be subscribable when a new customer is created or updated we can register an event as:
name- unique namelimetype- limetype the custom event applies toevents- events to register for
The above example will need an event to be published on core.limeobject.company.customer.new or core.limeobject.company.customer.update.
An example is to publish an event in a Company Custom LimeObject as:
from lime_type.unit_of_work import LimeobjectNewEvent
class NewCustomer(LimeobjectNewEvent):
def __init__(self, limeobject):
super().__init__(limeobject)
self.event_name = (
'core.limeobject'
f'.{self.limeobject.limetype.name}'
'.customer'
'.new.v1')
class Company(LimeObject):
def after_update(self, unsaved_self, **kwargs):
super().after_update(unsaved_self, **kwargs)
# Logic to check stuff
new_customer = NewCustomer(self)
new_customer.publish()