handlers.assign_tag_handler

class AssignTagHandler(handlers.eventhandler.EventHandler):

Assigns one or more tags whenever the configured trigger matches (see handlers/trigger.py). All gating lives in the trigger, so this handler covers any "tag the fact sheet if X" rule.

Configuration: trigger: When to assign the tags. Defaults to FactSheetCreatedEvent. tags: The tags to assign, as {"<tag group>": ["<tag>", ...]}. Already assigned tags are kept.

Example - tag every Application created by a specific user:

"trigger": {"fs_types": ["Application"], "event_match": {"userId": "<user id>"}},
"tags": {"Source": ["Created by technical user"]}

LeanIX system users share one user id and differ only by displayName - match those with creator_match (read from the fact sheet's first log entry), keeping event_match as the cheap pre-filter:

"trigger": {"fs_types": ["ITComponent"],
            "event_match": {"userId": "00000000-0000-4000-0000-000000000010"},
            "creator_match": {"displayName": "System user 'trc-data-digest'"}}

Initial sync: if the trigger matches on the event's userId or uses creator_match, the creator of each existing fact sheet is looked up from its log events (once), so the rule is applied retroactively as if the fact sheet had just been created. Otherwise the standard initial sync event is used.

name = 'Assign tag'

description = 'Assigns one or more tags whenever the configured trigger matches, e.g. if a fact sheet is created by a specific user.'

conditions = {'event_types': ['FactSheetCreatedEvent']}

supports_initial_sync = True

config_template = {'trigger': {'fs_types': ['Application'], 'event_match': {'userId': '00000000-0000-0000-0000-000000000000'}}, 'tags': {'Source': ['Created by technical user']}}

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 handle(self, event, fs, **kwargs):

Called automatically whenever a matching event comes in via webhook.

def initial_sync_event(self, fs):

Inherited Members

handlers.eventhandler.EventHandler: EventHandler; get_trigger; get_event_types; get_fs_types; get; all; lx_client; update_with_uniqueness_check; facet_matches; is_tag_key; SUBSCRIPTION_TYPES; parse_subscription_key; get_mapped_subscriptions; get_notification_users; get_send_email_for_users; get_email_callback; get_due_date