handlers.aggregation_across_relation

class AggregationAcrossRelation(handlers.eventhandler.EventHandler):

Aggregates information from child to parent.

Configuration:
relation: The relation across which you want to transfer information. Can be found in the Meta Model Configuration of your workspace. fields: List of fields that should be transferred. Can have one of the following attribute combinations:

name and priority: Transfers the named field across the relation, and if multiple fact sheets are linked, the priority is considered.
Example: Calculate the compliance of a parent application, and if one child application is non-compliant, the parent should be non-compliant as well.

name, source, mode and format: Can be used to calculate sums of fields on relations, e.g. cost fields on the relation are summed up on the fact sheet itself.
Example: Calculate the cost of an application by summing up the cost of its components.

name, source, mode and format: Can be used to join fields on relations or fact sheets using a separator character.
Example: Join the list of all properties of all associated fact sheets into a CSV value.

name, source, distribution_key, mode and format: Same as above, but calculates cost shares and considers the distribution key accordingly.
Example: Calculate the cost split of an application based on the user numbers of the organizations that use the application.

filter: Custom expression for determining the value list. Also supports multi-line functions.
Example 1 - only consider source fact sheets with a specific lifecycle or name: ["return [source_fs[src_field] for source_fs in fs.relation(target_to_source_relation) if source_fs[src_field] and (source_fs['lifecycle_status'] == 'active' or source_fs['name'] == 'Customers')]"]
Example 2 - only consider target fact sheets with a specific tag: "filter": ["return [source_fs[src_field] for source_fs in fs.relation(target_to_source_relation)] if fs.has_tag(tag='Local Application', tag_group='Category') else []"]

tags: Aggregates tags instead of fields. Can be configured using the following settings:
group: Name of the tag group
mode: "priority" or "aggregation". If "aggregation", all tags in the group are aggregated onto the target fact sheet.
priority: If mode is "priority", this setting specifies the priority, ranked highest to lowest. Only the highest-prioritized tag is transferred.
Example: If at least one of the child applications is Non-Compliant, the parent should get the "Non-Compliant" tag as well.

subscriptions: Aggregates subscriptions instead of tags or fields. Can be configured using the following settings:
subscription_type: Name of the subscription type, e.g. RESPONSIBLE
role: The name of the subscription role, e.g. "System Owner".
Example: All data owners from DataObject fact sheets are aggregated onto the applications that process those data objects.

name = 'Aggregation across Relation'

description = 'Aggregates information onto related fact sheets via specific rules.'

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

supports_initial_sync = True

config_template = {'relation': 'relDataObjectToApplication', 'fields': [{'name': 'security_compliance', 'priority': ['No', 'Yes']}, {'name': 'annualCost', 'source': 'rel:costTotalAnnual', 'mode': 'sum', 'format': '€ {:,.2f}'}, {'name': 'room', 'mode': 'join', 'separator': ';'}, {'name': 'permittedWriteACL', 'base': ['rbisg'], 'source': 'alias', 'mode': 'aggregation', 'split': ';'}, {'name': 'rel:costShare', 'source': 'annualCost', 'distribution_key': 'rel:numberOfUsers', 'mode': 'distribution', 'format': '€ {:,.2f}'}], 'tags': [{'group': 'security_compliance', 'mode': 'priority', 'priority': ['Non-Compliant', 'Compliant']}, {'group': 'security_characteristics', 'mode': 'aggregation'}], 'subscriptions': [{'subscription_type': 'RESPONSIBLE', 'role': 'System Owner'}]}

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 FactSheetUpdated event comes in via webhook.

def merge_lifecycles(self, lifecycles):

def max_lifecycle(self, lifecycles):

def get_fs_types(self):

Inherited Members

handlers.eventhandler.EventHandler: EventHandler; get_trigger; get_event_types; get; all; lx_client; initial_sync_event; 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