Write a function

Applies to VAST Cluster 5.5.

This guide covers the code inside a DataEngine function: the two functions your handler file defines, what each kind of event gives you, how a return value becomes the next function's event, how to read environment variables and secrets, and how to add Python packages. Each section shows the code and what it logged on a 5.5 cluster.

It builds on the quickstart: the thumbnails pipeline, the thumbnailer function and the docs-src-uploads trigger, and the environment variables set in its before you start section. The commands and outputs come from a lab cluster running VAST Cluster 5.5.0-SP6 with vastde v5.5.0-sp2. Your names, GUIDs, addresses, namespace and times will differ. For every class and method, see the Runtime SDK reference.

The handler file

vastde functions init python-pip <name> creates a project whose handler file is main.py. It defines two top-level functions:

init(ctx) runs once, when an instance of the function starts, before the instance receives any event. Use it for slow, one-time work: reading configuration, creating clients, loading a model. Attributes you set on ctx here are there in every later handler call on that instance.
handler(ctx, event) runs once per event. event is a VastEvent, a CloudEvent with properties for the fields DataEngine adds.

ctx gives you ctx.logger, ctx.secrets, metric instruments (ctx.counter() and others) and a tracer. See Context.

Both functions are required by the build, even if init has nothing to do. A handler file without init fails validation:

def handler(ctx, event):
    ctx.logger.info("got %s", event.id)

vastde functions build "$REGISTRY/docs/noinit" --target ./noinit --image-tag v1

Detected language: python
Handler file validation failed: noinit/main.py
Missing required components:
  - [async] def init(ctx) - Initialization function
Please fix the handler file and try again.
Handler file validation failed: noinit/main.py
Requirements:
  - [async] def init(ctx) - Initialization function

Sync or async

Both functions can be def or async def. The runtime logs which kind it found when the instance starts, for example Sync handler detected, concurrency limited to 20 via middleware.

A def handler runs in a pool of worker threads, so several events can be in your handler at once. Anything shared through ctx or module globals must be thread-safe.
An async def handler runs on the instance's event loop. Don't block in it (blocking I/O, time.sleep, heavy CPU work); use async libraries or asyncio.to_thread.

Read the event

event.type and event.subtype say what kind of event it is. What else you can read depends on the type. The full attribute lists are in events.

Element events

An element trigger delivers one event per object (or file) created, deleted or tagged. Read the object from event.bucket and event.object_key, as the quickstart handler does:

def handler(ctx, event):
    # An element trigger delivers one event per new object
    ctx.logger.info("new object: bucket=%s key=%s", event.bucket, event.object_key)

After an upload, vastde logs get thumbnails --since 10m includes:

2026-10-08 12:26:58.94 [thumbnailer] [INFO]  [user] new object: bucket=docs-src key=photos/cat.jpg

For a new object, event.type is Element and event.subtype is ElementCreated, although you configure the trigger with ObjectCreated:*. event.get_data() returns an S3-notification-style {"Records": [...]} with the object's size and ETag. Use event.bucket and event.object_key rather than parsing data; they work the same in local tests. See element events.

Schedule events

A schedule trigger delivers an event each time its cron expression comes due. This handler logs everything a schedule event carries:

import json

def init(ctx):
    ctx.logger.info("ticker initialized")

def handler(ctx, event):
    # Log everything a schedule trigger delivers
    ctx.logger.info("attributes: %s", json.dumps(dict(event.get_attributes()), default=str, sort_keys=True))
    data = event.get_data()
    ctx.logger.info("data (%s): %s", type(data).__name__, json.dumps(data, default=str, sort_keys=True))
    props = {}
    for name in ("id", "type", "subtype", "trigger", "trigger_id", "broker", "topic",
                 "partition_key", "extensions", "cron_schedule", "timer_elapsed_timestamp"):
        try:
            props[name] = getattr(event, name)
        except Exception as e:  # not every property applies to every event type
            props[name] = f"<{type(e).__name__}: {e}>"
    ctx.logger.info("properties (%s): %s", type(event).__name__, json.dumps(props, default=str, sort_keys=True))

One event, from a trigger with the schedule */2 * * * *, in vastde logs get ticks --scope user:

2026-10-08 13:00:01.08 [ticker] [INFO]  [user] attributes: {"cronschedule": "*/2 * * * *", "datacontenttype": "application/json", "dataschema": null, "id": "17914644000033660207", "knativekafkaoffset": "2", "knativekafkapartition": "3", "partitionkey": "17914644000033660207", "source": "vastdata.com:every-2-min.f846c90c-cbf4-4efa-9af8-1c9e546d7a75", "specversion": "1.0", "subject": "docs-kafka.default", "time": "2026-10-08T13:00:00.003000+00:00", "timerelapsedtimestamp": "2026-10-08T13:00:00.003361Z", "type": "vastdata.com:Schedule.TimerElapsed"}
2026-10-08 13:00:01.08 [ticker] [INFO]  [user] data (dict): {"message": "Activating trigger by cron"}
2026-10-08 13:00:01.08 [ticker] [INFO]  [user] properties (ScheduleVastEvent): {"broker": "docs-kafka", "cron_schedule": "*/2 * * * *", "extensions": {"cronschedule": "*/2 * * * *", "dataschema": null, "knativekafkaoffset": "2", "knativekafkapartition": "3", "partitionkey": "17914644000033660207", "timerelapsedtimestamp": "2026-10-08T13:00:00.003361Z"}, "id": "17914644000033660207", "partition_key": "17914644000033660207", "subtype": "TimerElapsed", "timer_elapsed_timestamp": "2026-10-08T13:00:00.003361Z", "topic": "default", "trigger": "every-2-min", "trigger_id": "f846c90c-cbf4-4efa-9af8-1c9e546d7a75", "type": "Schedule"}

The event's class is ScheduleVastEvent. Use event.timer_elapsed_timestamp (when the schedule fired) rather than the time the handler runs: delivery takes a few seconds. data is always the same fixed message, so there's nothing to read there. event.trigger names the trigger, which helps when one function serves several schedules. See triggers.

Function events

A function linked downstream of another function receives a Function event (class FunctionVastEvent) whose data is the upstream function's return value. See return values and chaining.

dict(event.get_attributes()) gives you the CloudEvent attributes as a dictionary; event.get_attributes() on its own isn't a plain dict.

Return values and chaining

What a handler returns isn't sent back to whoever delivered the event. If a link in the pipeline starts at this function, the return value becomes the data of a new event for the link's destinations. If no link starts there, it's discarded, and the runtime logs No sink URL configured.

This version of thumbnailer works out a thumbnail key and returns it:

import os

def init(ctx):
    # One-time initialization: runs once when the function instance starts
    ctx.logger.info("thumbnailer v2 initialized")

def handler(ctx, event):
    # An element trigger delivers one event per new object
    name = os.path.basename(event.object_key)
    thumbnail_key = f"thumbnails/{name}"
    ctx.logger.info("v2: thumbnail for s3://%s/%s -> %s", event.bucket, event.object_key, thumbnail_key)
    # The return value becomes the event for any function linked downstream
    return {"bucket": event.bucket, "key": event.object_key, "thumbnail_key": thumbnail_key}

A second function, indexer, logs what it receives:

import json

def init(ctx):
    ctx.logger.info("indexer initialized")

def handler(ctx, event):
    # Log everything the function receives from the upstream function
    ctx.logger.info("attributes: %s", json.dumps(event.get_attributes(), default=str, sort_keys=True))
    data = event.get_data()
    ctx.logger.info("data (%s): %s", type(data).__name__, json.dumps(data, default=str, sort_keys=True))
    props = {}
    for name in ("id", "type", "subtype", "trigger", "trigger_id", "broker", "topic",
                 "partition_key", "extensions", "function_trigger"):
        try:
            props[name] = getattr(event, name)
        except Exception as e:  # not every property applies to every event type
            props[name] = f"<{type(e).__name__}: {e}>"
    ctx.logger.info("properties (%s): %s", type(event).__name__, json.dumps(props, default=str, sort_keys=True))
    ctx.logger.info("indexed %s (thumbnail %s)", data["key"], data["thumbnail_key"])

In the pipeline file, a second link sends thumbnailer's output to indexer:

  links:
    - source: [uploads]
      destination: [thumbnailer]
    - source: [thumbnailer]
      destination: [indexer]

Pipelines as code has the full file and the update and deploy commands. After an upload, the logs show the hand-off:

vastde logs get thumbnails --since 1m

…
2026-10-08 12:45:01.77 [thumbnailer] [INFO]  [user] v2: thumbnail for s3://docs-src/photos/bee.jpg -> thumbnails/bee.jpg
2026-10-08 12:45:01.77 [thumbnailer] [INFO]  [vast-runtime] Function returned: {'bucket': 'docs-src', 'key': 'photos/bee.jpg', 'thumbnail_key': 'thumbnails/bee.jpg'} in timestamp 2026-10-08 12:45:01.779760 with duration 0.5428791046142578 milliseconds
2026-10-08 12:45:01.78 [thumbnailer] [INFO]  [vast-runtime] sending CloudEvent to sink URL: http://kafka-broker-ingress.knative-eventing.svc.cluster.local/dataengine-ns-da47bae0-aefd-4e85-a5a9-2020d6a2cb91/docs-kafka-058da6a90bb18d99f58e
2026-10-08 12:45:01.93 [thumbnailer] [INFO]  [vast-runtime] CloudEvent sent to sink: http://kafka-broker-ingress.knative-eventing.svc.cluster.local/dataengine-ns-da47bae0-aefd-4e85-a5a9-2020d6a2cb91/docs-kafka-058da6a90bb18d99f58e in timestamp 2026-10-08 12:45:01.939119
…
2026-10-08 12:45:01.95 [indexer] [INFO]  [vast-runtime] START EventId: 87982405058564, TraceId: -, EventType: vastdata.com:Function, EventSource: vastdata.com:thumbnailer-nbdn4ypiyaaaa, Timestamp: 1791463501.956112
…
2026-10-08 12:45:01.95 [indexer] [INFO]  [user] attributes: "{'source': 'vastdata.com:thumbnailer-nbdn4ypiyaaaa', 'id': '87982405058564', 'type': 'vastdata.com:Function', 'specversion': '1.0', 'time': '2026-10-08T12:45:01.778000+00:00', 'subject': 'vast-broker-docs-kafka.default', 'datacontenttype': 'application/json', 'dataschema': None, 'knativekafkaoffset': '0', 'knativekafkapartition': '7', 'partitionkey': 'docs-src/photos/bee.jpg'}"
2026-10-08 12:45:01.95 [indexer] [INFO]  [user] data (dict): {"bucket": "docs-src", "key": "photos/bee.jpg", "thumbnail_key": "thumbnails/bee.jpg"}
2026-10-08 12:45:01.95 [indexer] [WARN]  [user] vast.dataengine.sdk.event: subtype is not available for function event type
2026-10-08 12:45:01.95 [indexer] [WARN]  [user] vast.dataengine.sdk.event: Internal function event does not have a trigger, use function_trigger property instead
2026-10-08 12:45:01.95 [indexer] [WARN]  [user] vast.dataengine.sdk.event: Internal function event does not have a trigger_id, use function_trigger property instead
2026-10-08 12:45:01.95 [indexer] [INFO]  [user] properties (FunctionVastEvent): {"broker": "vast-broker-docs-kafka", "extensions": {"dataschema": null, "knativekafkaoffset": "0", "knativekafkapartition": "7", "partitionkey": "docs-src/photos/bee.jpg"}, "function_trigger": "thumbnailer-nbdn4ypiyaaaa", "id": "87982405058564", "partition_key": "docs-src/photos/bee.jpg", "subtype": null, "topic": "default", "trigger": null, "trigger_id": null, "type": "Function"}
2026-10-08 12:45:01.95 [indexer] [INFO]  [user] indexed photos/bee.jpg (thumbnail thumbnails/bee.jpg)
…

What this shows:

The data arrives as you returned it. indexer gets the dictionary back from event.get_data(). Return values that serialise to JSON (a dict, list, str, number or None) are safe.
The event keeps the original event's id and gets a partitionkey with the original object's path, so you can correlate the two. Its source names the upstream function's service (thumbnailer-nbdn4ypiyaaaa), not the trigger.
Trigger properties don't apply. On a function event, event.trigger, event.trigger_id and event.subtype return None, and the SDK logs a warning each time you read them. Read event.function_trigger instead, or don't read them.
The runtime logs every return value (the Function returned: line), and the value travels as event data. Never return a secret.
None is still sent on as an event with no data. To send events down only some links, use conditional trigger labels; see conditional routing.

Errors

To fail an event, raise an exception. The runtime logs it, marks the trace as an error, and DataEngine retries the delivery; when the retries run out, the event goes to the tenant's dead-letter topic. A retry runs your handler again with the same event, so make handlers safe to run twice: derive output names from the input, and overwrite rather than append.

def init(ctx):
    ctx.logger.info("faulty initialized")

def handler(ctx, event):
    ctx.logger.info("processing %s", event.object_key)
    raise ValueError(f"cannot process {event.object_key}")

With this function in a pipeline called errors-demo, one upload under fail/ gives:

vastde logs get errors-demo --since 30m --severity ERROR --limit 5

2026-10-08 12:53:07.18 [faulty] [ERROR] [vast-runtime] Error in handler execution: cannot process fail/broken.txt
2026-10-08 12:53:09.19 [faulty] [ERROR] [vast-runtime] Error in handler execution: cannot process fail/broken.txt
2026-10-08 12:53:13.21 [faulty] [ERROR] [vast-runtime] Error in handler execution: cannot process fail/broken.txt
2026-10-08 12:53:21.22 [faulty] [ERROR] [vast-runtime] Error in handler execution: cannot process fail/broken.txt
2026-10-08 12:53:37.23 [faulty] [ERROR] [vast-runtime] Error in handler execution: cannot process fail/broken.txt

Each line is one attempt at the same event, with a growing wait between them.

To skip an event without a retry, catch the exception, log it and return normally. Retries and the dead-letter topic shows the retries, how to set their number, and how to read the dead-letter topic.

Environment variables and secrets

Use environment variables for settings that change between deployments and aren't sensitive (bucket names, sizes, a stage name), and secrets for anything sensitive. You set both in the pipeline file, not in the image, so one image runs everywhere.

This handler reads three environment variables and a secret. It logs only the secret's name and length, never its value:

import os

def init(ctx):
    ctx.logger.info("envcheck initialized")

def handler(ctx, event):
    # Environment variables: pipeline-level and function-level
    for name in ("STAGE", "THUMBNAIL_SIZE", "LOG_PREFIX"):
        ctx.logger.info("env %s=%r", name, os.environ.get(name))
    # Secrets: log only secret and key names, never values
    ctx.logger.info("secrets: %s", {name: sorted(keys) for name, keys in ctx.secrets.items()})
    token = ctx.secrets["thumbnail-api"]["api-token"]
    ctx.logger.info("api-token is %d characters long", len(token))

ctx.secrets is a dictionary of secrets by name, each a dictionary of its keys and values. Build and create this function as envcheck, as in the quickstart, and create an element trigger docs-src-env for keys that start with env/.

Create the secret

A secret file names the secret, the cluster and namespace it lives in, and its entries. Save this as thumbnail-api-secret.yaml, with your cluster, namespace and token, and keep it out of Git:

name: thumbnail-api
kubernetes_cluster_vrn: vast:dataengine:kubernetes-clusters:test-cluster
namespace: dataengine-ns-da47bae0-aefd-4e85-a5a9-2020d6a2cb91
entries:
  - key: api-token
    value: "<your API token>"

Set them in the pipeline

Environment variables go under config.environment_variables, for the whole pipeline (manifest.config) or for one function deployment. Secrets are attached by name under config.secrets. Save this as pipeline.yaml in the envcheck project's parent directory:

name: env-demo
description: Environment variables and secrets
kubernetes_cluster_vrn: vast:dataengine:kubernetes-clusters:test-cluster
namespace: dataengine-ns-da47bae0-aefd-4e85-a5a9-2020d6a2cb91
manifest:
  config:
    environment_variables:
      - name: STAGE
        value: docs
      - name: THUMBNAIL_SIZE
        value: "128"
  triggers:
    - name: env-uploads
      vrn: vast:dataengine:triggers:docs-src-env
  function_deployments:
    - name: envcheck
      function_vrn: vast:dataengine:functions:envcheck
      config:
        environment_variables:
          - name: THUMBNAIL_SIZE
            value: "256"
          - name: LOG_PREFIX
            value: envcheck
        secrets:
          - thumbnail-api
  links:
    - source: [env-uploads]
      destination: [envcheck]

Values are strings: quote numbers. --secret-file creates the secret before the pipeline:

vastde pipelines create --config @pipeline.yaml --secret-file thumbnail-api-secret.yaml --deploy

Creating secret: thumbnail-api
Secret created: thumbnail-api
Pipeline created: env-demo
Name: env-demo
Description: Environment variables and secrets
Tags: []
GUID: 683ecd67-2751-4400-84e5-f5d0286ca93d
Owner: [id: 4294967295, id-type: vid]
Created At: 2026-10-08T13:01:34Z
Updated At: 2026-10-08T13:01:34Z
VRN: vast:dataengine:pipelines:env-demo

Deploying pipeline: env-demo
Pipeline deployed successfully

After an upload that fires the trigger:

vastde logs get env-demo --since 2m --scope user

2026-10-08 13:02:05.33 [envcheck] [INFO]  [user] envcheck initialized
2026-10-08 13:02:56.73 [envcheck] [INFO]  [user] env STAGE='docs'
2026-10-08 13:02:56.73 [envcheck] [INFO]  [user] env THUMBNAIL_SIZE='256'
2026-10-08 13:02:56.73 [envcheck] [INFO]  [user] env LOG_PREFIX='envcheck'
2026-10-08 13:02:56.73 [envcheck] [INFO]  [user] secrets: {'thumbnail-api': ['api-token']}
2026-10-08 13:02:56.73 [envcheck] [INFO]  [user] api-token is 48 characters long

STAGE comes from the pipeline level. THUMBNAIL_SIZE is set at both levels, and the function deployment's value, 256, wins.

For updating secrets and keeping them out of logs, images and Git, see security.

What the cluster rejects

Some names are reserved for DataEngine and the runtime. Setting one fails the create (or update), and the message lists them:

vastde pipelines create --config @reserved-env.yaml --deploy

pipeline creation failed: The following environment variable names are reserved for internal use and cannot be set by users: VAST_ENDPOINT. Reserved prefixes: VASTPIPELINE_, K_, VAST_. Reserved names: HOSTNAME, PERFORMANCE, PORT, VAST_RUNTIME_HANDLER_THREADS

reserved-env.yaml set VAST_ENDPOINT at the pipeline level; PORT and K_SINK give the same error. So don't use the VAST_ENDPOINT, VAST_ACCESS_KEY and similar names for database functions: use your own prefix, such as DB_ENDPOINT, and keep keys in a secret.

A secret can be attached to the whole pipeline or to a function deployment, not both:

vastde pipelines create --config @secret-twice.yaml --deploy

pipeline creation failed: Secret(s) already defined at pipeline level cannot be redefined at function deployment level. Duplicates found in: 'envcheck': thumbnail-api. Remove the duplicate from either the pipeline or the function deployment config.

A secret that doesn't exist isn't caught when you create the pipeline, only when you deploy it:

vastde pipelines create --config @missing-secret.yaml --deploy

Pipeline created: env-errors
Name: env-errors
Tags: []
GUID: b710c8ee-85fd-4f2b-87dd-a7eaf58ef06a
Owner: [id: 4294967295, id-type: vid]
Created At: 2026-10-08T13:04:00Z
Updated At: 2026-10-08T13:04:00Z
VRN: vast:dataengine:pipelines:env-errors

Deploying pipeline: env-errors
pipeline deployment failed: Failed to deploy pipeline. Pipeline: 'env-errors' (Kubernetes name: 'env-errors', vast-id: 'nbdoorbbiaaaa', GUID: b710c8ee-85fd-4f2b-87dd-a7eaf58ef06a). Hint: Required secrets are missing. Ensure all secrets referenced in the pipeline manifest exist in the target namespace.

The pipeline exists but isn't deployed. Create the secret (with --secret-file on vastde pipelines update) and deploy again, or delete the pipeline.

Dependencies

The scaffold has three places for what your code needs. The builder installs them into the image when you run vastde functions build:

File	What goes in it
`requirements.txt`	Python packages, one per line, as for `pip install -r`. The build installs them from PyPI (or your mirror).
`Aptfile`	System packages (Ubuntu), one per line, for libraries your Python packages need.
`customDeps`	Paths to local libraries (a directory, a file or a zip) that the build copies into the image's Python path.

You don't add the DataEngine SDK or runtime: the builder adds them.

A package your code imports but that isn't listed doesn't fail the build; it fails when the function starts. This handler imports Pillow without listing it:

from PIL import Image  # Pillow is not listed in requirements.txt

def init(ctx):
    ctx.logger.info("badimport initialized")

def handler(ctx, event):
    ctx.logger.info("Pillow %s", Image.__version__)

The build, the function and the pipeline are all created without complaint, but the pipeline doesn't become Ready. It stays InProgress for several minutes and then goes to Error, and vastde logs get has nothing, because the process exits before it can send any logs:

vastde logs get errors-init --since 10m

No logs found. Try increasing the start time or changing the scope.

The traceback is in the pipeline's reason, once its status is Error:

vastde pipelines get errors-init -o yaml

…
reason: |-
    Subresource failed: Function 'badimport' deployment failed: RevisionFailed - Revision "badimport-nbdou5keyaaaa-00001" failed with message: Container failed with: 2026-10-08 13:25:02 [INFO] [trace_id=-] VAST DataEngine Runtime, Version: 0.2.0
    …
      File "/workspace/main.py", line 1, in <module>
        from PIL import Image  # Pillow is not listed in requirements.txt
        ^^^^^^^^^^^^^^^^^^^^^
    ModuleNotFoundError: No module named 'PIL'
    .
status: Error
…

You'd see the same traceback within seconds by running the image locally first: see build and test locally. Add the package to requirements.txt, build a new image tag and publish it.