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.

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))

Register the storage location

For a cloud bucket or container, first connect the storage location and configure provider delivery. This creates the Tilebox location and connects S3/SNS, GCS/Pub/Sub, or Azure Blob/Event Grid notifications. Registering an automation alone does not configure cloud notifications or grant the runner object-read access.

Select the registered location that the automation should watch:

from tilebox.workflows import Client

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

Register the automation

Create the automation using the task prototype and storage trigger.

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

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>

Next steps