handlers.eventhandler API documentation

EventHandler(config, lx_client=None, **kwargs)

conditions = {'fs_types': [], 'event_types': []}

supports_initial_sync = False

def get_trigger(self):

def get_event_types(self):

def get_fs_types(self):

@classmethod

def get(cls, name):

@classmethod

def all(cls):

lx_client

def handle(self, event, fact_sheet, **kwargs):

Stub, implemented by actual handlers (child classes)

def data_requirements(self):

Declare the exact fact-sheet data this handler needs, so the webhook event loop (process_event in workspace/views_module.py) can load a single lean FastFactSheet per event instead of the full fact sheet.

Return a dict: { "fields": [], "relations": [], "subscriptions": bool, # True if handle() reads fs["subscriptions"] "tags": bool, # True if handle() reads tags / has_tag() "documents": bool, "acls": bool, "completion": bool, # True if handle() reads fs["completion"] "base_fields": [], # optional; omit/None = full # default base field set. A list restricts the # base scalars to just those (id and type are # always loaded). Advanced opt-in for max speed. } Fields referenced by this handler's trigger are folded in automatically by the event loop (into "fields"), so only declare what handle() itself needs on top of them.

Return None (the default) to signal "requirements cannot be declared statically" - e.g. handlers that exec/eval config-supplied code. The event loop then loads the full fact sheet for that event, so correctness is always preserved. This is a standalone hook (deliberately not tied to the initial_sync_* hooks, which are slated to migrate onto it later).

def initial_sync_event(self, fs):

def update_with_uniqueness_check(self, fs, field, value, unique_id_field):

def facet_matches(self, facet, fs):

def is_tag_key(self, facet_key):

SUBSCRIPTION_TYPES = ('RESPONSIBLE', 'ACCOUNTABLE', 'OBSERVER')

@classmethod

def parse_subscription_key(cls, key):

Parses a SUBSCRIPTION mapping key "subscriptions//" into (subscription_type, role).

is RESPONSIBLE, ACCOUNTABLE or OBSERVER (case-insensitive). Empty, "*" or "ANY" means any type, so the common legacy form "subscriptions//" keeps matching the role regardless of type. Any other value is ignored with a warning (legacy configs used it as a free-text label).
is everything after the second slash, so role names containing "/" work as-is ("subscriptions/RESPONSIBLE/Sales/Marketing Owner"). An escaped "\/" is accepted as well. An empty role selects the role-less subscriptions of that type.
The legacy two-part form "subscriptions/" matches the role with any type.

@classmethod

def get_mapped_subscriptions(cls, fs, key, subscription_filter=None):

Returns the subscriptions of fs selected by a SUBSCRIPTION mapping key, see parse_subscription_key().

subscription_filter: optional dict of subscription attribute -> regular expression, e.g. {"comment": "IT-Architect"} (a mapping's "subscriptionFilter"). Only subscriptions whose attributes all match (re.search; a missing/None value is matched as "") are returned - if none match, none are returned.

def get_notification_users(self, fs, event, roles, user_filter=None):

user_filter: optional callable(user) -> bool. Users for which it returns False are skipped as if they were not subscribed, so that __FIRST_MATCHING__ falls through to the next role.

def get_send_email_for_users(self, notification_users, fs, rel, errors, due, title, description):

def get_email_callback(self, notification_users, fs, rel, errors, due, title, description):

Create a callback for sending an email to users when TODOs are created/changed for them.

def get_due_date(self, fs):