--- title: "Write a function" slug: "dev-dataengine-guides-write-a-function" description: "Explore DataEngine functions, event handling, environment variables, and Python package integration in this comprehensive guide for VAST Cluster 5.5." tags: ["DataEngine"] updated: 2026-10-28T02:14:59Z published: 2026-10-28T02:14:59Z canonical: "docs.vastdata.com/dev-dataengine-guides-write-a-function" --- > ## Documentation Index > Fetch the complete documentation index at: https://docs.vastdata.com/llms.txt > Use this file to discover all available pages before exploring further. # 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](/develop/docs/dev-dataengine-quickstart): the `thumbnails` pipeline, the `thumbnailer` function and the `docs-src-uploads` trigger, and the environment variables set in its [before you start](/develop/docs/dev-dataengine-quickstart#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](/develop/docs/dev-dataengine-reference-runtime-sdk). ## 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](/develop/docs/dev-dataengine-reference-runtime-sdk#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](/develop/docs/dev-dataengine-reference-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](/develop/docs/dev-dataengine-reference-events#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](/develop/docs/dev-dataengine-guides-triggers#schedule-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](/develop/docs/dev-dataengine-guides-write-a-function#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](/develop/docs/dev-dataengine-guides-pipelines-as-code#add-a-function-to-the-chain) 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](/develop/docs/dev-dataengine-guides-pipelines-as-code#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](/develop/docs/dev-dataengine-guides-pipelines-as-code#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](/develop/docs/dev-dataengine-quickstart#3-build-and-push-the-image), 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: "" ``` ### 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](/develop/docs/dev-dataengine-operations-security#secrets). ### 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 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](/develop/docs/dev-dataengine-guides-build-and-test-locally#catch-start-up-errors). Add the package to `requirements.txt`, build a new image tag and [publish it](/develop/docs/dev-dataengine-guides-publish-functions). ## See also - [Runtime SDK reference](/develop/docs/dev-dataengine-reference-runtime-sdk): `Context`, `VastEvent`, batch handlers, tracing and metrics. - [Events](/develop/docs/dev-dataengine-reference-events): every attribute of each event type. - [Build and test locally](/develop/docs/dev-dataengine-guides-build-and-test-locally) - [Observability](/develop/docs/dev-dataengine-guides-observability): logs, traces and custom metrics. - KB: [Runtime SDK Developer's Guide](/docs-5-5/docs/vast-cluster-5-5-dataengine-runtime-sdk-developer-s-guide)