Storage Event Triggers

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

class LogObjectCreation(StorageEventTask):
    head_bytes: int

    def execute(self, context: ExecutionContext) -> None:
        # self.trigger is an attribute of the StorageEventTask class,
        # which contains information about the storage event that triggered this task
        if self.trigger.type == StorageEventType.CREATED:
            path = self.trigger.location
            context.logger.info("A new object was created", path=path)

            # trigger.storage is a storage client for interacting with the storage
            # location that triggered the event
            # using it, we can read the object to get its content as a bytes object
            data = self.trigger.storage.read(path)
            context.logger.info("Read object data", path=path, size_bytes=len(data))
            context.logger.info("Object preview", path=path, preview=data[:self.head_bytes].hex())

type LogObjectCreation struct {
    HeadBytes int
}

func (t *LogObjectCreation) Execute(ctx context.Context) error {
    slog.InfoContext(ctx, "storage event task triggered", slog.Int("head_bytes", t.HeadBytes))
    return nil
}

from tilebox.workflows import Client

client = Client()
automations_client = client.automations()

storage_locations = automations_client.storage_locations()
print(storage_locations)

ctx := context.Background()
client := workflows.NewClient()

storageLocations, err := client.Automations.ListStorageLocations(ctx)
if err != nil {
    slog.ErrorContext(ctx, "failed to list storage locations", slog.Any("error", err))
    return
}

for _, location := range storageLocations {
    slog.InfoContext(ctx,
        "storage location",
        slog.String("id", location.ID.String()),
        slog.String("location", location.Location),
        slog.String("type", location.Type.String()),
    )
}

from tilebox.workflows import Client, Runner

client = Client()
runner = Runner(tasks=[LogObjectCreation])
runner.connect_to(client).run_forever()

ctx := context.Background()
client := workflows.NewClient()

runner, err := client.NewTaskRunner(ctx)
if err != nil {
    slog.ErrorContext(ctx, "failed to create task runner", slog.Any("error", err))
    return
}

if err := runner.RegisterTasks(&LogObjectCreation{}); err != nil {
    slog.ErrorContext(ctx, "failed to register tasks", slog.Any("error", err))
    return
}

if err := runner.RunForever(ctx); err != nil {
    slog.ErrorContext(ctx, "task runner stopped", slog.Any("error", err))
}