Event Handling

The handler function is invoked whenever the function receives an event or a batch of events. Events can be passed to each function as single or batched events, depending whether batching is configured in the function deployment configured in the pipeline.

The runtime supports asynchronous event handling, which is recommended for improved throughput and performance. You can define your event handler as async def.

For processing single events:

async def handler(ctx, event)

For processing a batch of events:

async def handler(ctx, events_list)

The return value can be either of:

Import the class and subclasses into your code from vast.dataengine.sdk.

For example:

from vast.dataengine.sdk import Context, VastEventList, event_batching_handler

You can cast event to a specific subclass to access properties that are specific to the event type:

 try:
        schedule_event = event.as_schedule_event()
        ctx.logger.info(f"Schedule cron: {schedule_event.cron_schedule}")
        ctx.logger.info(f"Schedule timer elapsed: {schedule_event.timer_elapsed_timestamp}")
    except TypeError:
            ctx.logger.info("validate-event: expected schedule event")

The following abstractions help you implement batch event processing in a function:

To implement batch event processing:

Example:

from typing import List

#  Import the necessary classes and decorators from the DataEngine SDK:
from vast.dataengine.sdk import Context, VastEventList, event_batching_handler

# Initialize the DataEngine runtime when a function instance starts up.
def init(ctx: Context):
      # Use the SDK's context logger to write an informational log indicating the function instance has been initialized.
      ctx.logger.info("Initializing Bulk Event Processor...")

# Tell DataEngine that this function is configured to receive event batches.
@event_batching_handler
# Pass events_list as the second argument in the event handler, to accept batches of events wrapped in VastEventList. DataEngine accumulates events in batches according to batch size and batch timeout configured in function deployment. 
def handler(ctx: Context, events_list: VastEventList):
    ctx.logger.info(f"Processing batch of {len(events_list)} events in '{ctx.function_name}'.")

    # Extract data dictionaries from the batch wrapper
    payloads = [event.get_data() for event in events_list]

    # Execute bulk operation (e.g., 1 database request instead of N requests)
    perform_bulk_operation(payloads)

    # Mutate payloads if necessary and return the list
    for payload in payloads:
        payload["processed_by"] = ctx.function_name

    # Return list of processed payloads to send downstream
    return payloads


def perform_bulk_operation(records: List[dict]):
    """
    Placeholder helper function for bulk DB insertions or batch API calls.
    Replace with actual database or HTTP client call.
    """
    # e.g., db.execute_many("INSERT INTO events VALUES (...)", records)
    pass

To implement conditional function routing, use the set_trigger_labels() setter method to attach labels to event objects. In your DataEngine pipelines, if you attach a label to a connector that routes the function to another function, only events that have a label matching the configured trigger label will invoke the function.

Note

Passing an empty dictionary to event.set_trigger_labels({}) is not allowed and raises a ValueError. To intentionally prevent an event from routing downstream, attach a dummy label that does not match any configured connector (e.g., {"noRoute": "skip"}).

For example, suppose you have an e-commerce platform processing customer orders. When a new order occurs, the incoming order is received as an event. Depending on what is inside the order, the pipeline needs to route the data to different specialized downstream function. You might have all events routed to an initial sorting function which routes the orders to different functions downstream.

You might use conditional function routing to implement this in your pipeline. To do this, you would write your first function, function A, to inspect the payload fields, construct routing labels and attach them to the events before returning them to the pipeline. In the pipeline, you connect the trigger that produces the events to Function A, which would look something like this:

import json
from vast.dataengine.sdk import VastEvent

def init(ctx):
    pass

def handler(ctx, event: VastEvent):
    data = event.get_data()
    message_str = data["message"]
    
    # 1. Parse the incoming raw JSON message into a structured dictionary
    payload = json.loads(message_str)
    
    # 2. Create an empty dictionary to hold trigger labels
    labels = {}
    
    # 3. Apply business logic to build labels from raw payload fields
    customer_tier = payload.get("customer", {}).get("tier")
    if customer_tier == "VIP":
        labels["tier"] = "vip_customer"
        
    risk_score = payload.get("payment", {}).get("risk_score", 0)
    if risk_score > 80:
        labels["alert_type"] = "security_flag"
        
    order_amount = payload.get("payment", {}).get("amount", 0)
    if order_amount >= 1000:
        labels["value_tier"] = "high_value"

    # 4. Apply the extracted/computed labels to the event
    if labels:
        ctx.logger.info(f"Applying generated trigger labels: {labels}")
        event.set_trigger_labels(labels)
    else:
        # Handle non-matching routing (must use non-empty dict)
        event.set_trigger_labels({"noRoute": "skip"})
        
    # 5. Return the structured dict payload for downstream pipeline steps
    return payload

You might then prepare function B to process events labelled vip_customer, function C to process events labelled security_flag, and function D to process events labelled high_value.

On the DataEngine platform, when building the pipeline, you would route function A to functions B, C and D, and apply event trigger labels on the connectors from function A to each of the other functions:

When multiple key-value pairs are configured on a routing link between functions, all labels must match the event in order for it to be routed to the linked function.

Successive calls to event.set_trigger_labels() on the same event object merge new key-value pairs into the existing label set rather than overwriting the whole dictionary. Existing keys are updated and new keys are added.This behavior is essential for fan-out routing, allowing different conditional checks in your function to attach distinct routing labels incrementally without clearing previously set labels.

# Step 1: Attach initial routing label based on user tier
event.set_trigger_labels({"tier": "vip"})

# Step 2: Attach an additional label based on region
# This merges with step 1 rather than replacing it
event.set_trigger_labels({"region": "us-east"})

# Final active label set on event:
# {"tier": "vip", "region": "us-east"}
# Downstream connectors matching either "tier=vip" OR "region=us-east" will trigger.

Conditional routing operates strictly hop-by-hop along declared pipeline links: