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()