Configure storage-event automations

Use storage-event automations when new or changed files should trigger workflow work, such as processing new uploads in a bucket or reacting to generated products.

Tilebox submits a job when a matching storage event occurs. A runner on the target cluster still needs to execute the submitted task.

Define the storage-event task

Create a task that handles the event payload.

Python

from tilebox.workflows import ExecutionContext
from tilebox.workflows.automations import StorageEventTask, StorageEventType


class ProcessUploadedObject(StorageEventTask):
    head_bytes: int = 64

    def execute(self, context: ExecutionContext) -> None:
        if self.trigger.type == StorageEventType.CREATED:
            path = self.trigger.location
            context.logger.info(
                "Processing uploaded object",
                path=path,
            )
            data = self.trigger.storage.read(path)
            context.logger.info("Read object data", size_bytes=len(data))

Copy

Register the storage location

Register or select the storage location that Tilebox should watch. Storage locations can represent cloud buckets or local file-system locations supported by Tilebox automations.

Python

from tilebox.workflows import Client

client = Client()
automations = client.automations()
locations = automations.storage_locations()
print(locations)

Copy

Register the automation

Create the automation using the task prototype and storage trigger.

Python

automations.create_storage_event_automation(
    "process-uploaded-objects",
    ProcessUploadedObject(),
    triggers=[(locations[0], "incoming/**")],
)

Copy

Test and inspect

Create or change a matching object, then inspect the submitted job in the Console or through the Tilebox command-line tool.

tilebox job logs <job-id>
tilebox job spans <job-id>

Copy

Next steps

Storage event triggers

Read the full storage-event automation reference.

Debug a failed workflow run

Inspect task state, logs, traces, and runner context.