# Storage Event Triggers

## Creating a Storage Event Task

Storage Event Tasks are automations triggered when objects are created or modified in a [storage location](#storage-locations).
In Python, use `tilebox.workflows.automations.StorageEventTask` as your task base class instead of the regular `tilebox.workflows.Task`. In Go, register an executable task with the same task identifier as the automation prototype.

**Python**

```python title="Python"
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())
```

**Go**

```go title="Go"
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
}
```

## Storage Locations

Storage Event tasks are triggered when objects are created or modified in a storage location. This location can be a cloud storage bucket or a local file system. Tilebox supports the following storage locations:

**Google Cloud Storage**

**Amazon S3**

**Local File System**

### Registering a Storage Location

To make a storage location available within Tilebox workflows, it must be registered first. This involves specifying the location and setting up a notification system that forwards events to Tilebox, enabling task triggering. The setup varies depending on the storage location type.

For example, a GCP storage bucket is integrated by setting up a [PubSub Notification with a push subscription](https://cloud.google.com/storage/docs/pubsub-notifications). A local file system requires installing a filesystem watcher. To set up a storage location registered with Tilebox, please [get in touch](mailto:engineering@tilebox.com).

### Listing Available Storage Locations

To list all available storage locations, use the automation client.

**Python**

```python title="Python"
from tilebox.workflows import Client

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

storage_locations = automations_client.storage_locations()
print(storage_locations)
```

**Go**

```go title="Go"
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()),
    )
}
```

```plaintext title="Output"
[
    StorageLocation(
        location="gcp-project:gcs-bucket-fab3fa2",
        type=StorageType.GCS,
    ),
    StorageLocation(
        location="s3-bucket-263af15",
        type=StorageType.S3,
    ),
    StorageLocation(
        location='/path/to/a/local/folder',
        type=StorageType.FS,
    ),
]
```

### Reading Files from a Storage Location

Once a storage location is registered, you can read files from it using the `read` method on the storage client.

```python title="Python"
gcs_bucket = storage_locations[0]
s3_bucket = storage_locations[1]
local_folder = storage_locations[2]

gcs_object = gcs_bucket.read("my-object.txt")
s3_object = s3_bucket.read("my-object.txt")
local_object = local_folder.read("my-object.txt")
```

The `read` method instantiates a client for the specific storage location. This requires that
the storage location is accessible by a runner and may require credentials for cloud storage
or physical/network access to a locally mounted file system.

To set up authentication and enable access to a GCS storage bucket, check out the [Google Client docs for authentication](https://cloud.google.com/docs/authentication/client-libraries#python).

## Registering a Storage Event Trigger

After implementing a Storage Event task, register it to trigger each time a storage event occurs. The Python SDK provides a registration helper, and you can also register storage-event automations from the Tilebox Console.
This registration submits a new job consisting of a single task instance derived from the registered Storage Event task prototype.

```python title="Python"
from tilebox.workflows import Client

client = Client()
automations = client.automations()
storage_event_automation = automations.create_storage_event_automation(
    "log-object-creations",  # name of the storage event automation
    LogObjectCreation(head_bytes=20),  # the task (and its input parameters) to run repeatedly
    triggers=[
        # you can specify a glob pattern:
        # run every time a .txt file is created anywhere in the gcs bucket
        (gcs_bucket, "**.txt"),  
    ],
)
```

The syntax for specifying glob patterns follows [Standard Wildcards](https://tldp.org/LDP/GNU-Linux-Tools-Summary/html/x11655.htm).
Additionally, you can use `**` as a super-asterisk, a matching operator not sensitive to slash separators.

Here are some examples of valid glob patterns:

| Pattern     | Matches                                                                      |
| ----------- | ---------------------------------------------------------------------------- |
| `*.ext`     | Any file ending in `.ext` in the root directory                              |
| `**/*.ext`  | Any file ending in `.ext` in any subdirectory, but not in the root directory |
| `**.ext`    | Any file ending in `.ext` in any subdirectory, including the root directory  |
| `folder/*`  | Any file directly in a `folder` subdirectory                                 |
| `folder/**` | Any file directly or recursively part of a `folder` subdirectory             |
| `[a-z].txt` | Matches `a.txt`, `b.txt`, etc.                                               |

## Start a Storage Event runner

With the Storage Event automation registered, a job is submitted whenever a storage event occurs. But unless a [runner](/docs/workflows/concepts/runners) is available to execute the Storage Event task the submitted jobs remain in a task queue.
Once an [eligible runner](/docs/workflows/concepts/runners#task-selection) becomes available, all jobs in the queue are executed.

**Python**

```python title="Python"
from tilebox.workflows import Client, Runner

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

**Go**

```go title="Go"
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))
}
```

### Triggering an Event

Creating an object in the bucket where the task is registered results in a job being submitted:

```bash title="Creating an object"
echo "Hello World" > my-object.txt
gcloud storage cp my-object.txt gs://gcs-bucket-fab3fa2
```

Inspecting the runner output reveals that the job was submitted and the task executed:

```plaintext title="Output"
2024-09-25 16:51:45,621 INFO A new object was created: my-object.txt
2024-09-25 16:51:45,857 INFO The object's file size is 12 bytes
2024-09-25 16:51:45,858 INFO The object's first 20 bytes are: b'Hello World\n'
```

## Inspecting in the Console

The [Tilebox Console](https://console.tilebox.com/workflows/automations) provides an easy way to inspect all registered storage event automations.

![Tilebox Workflows automations in the Tilebox Console](/docs/assets/console/storage-event-light.png)

![Tilebox Workflows automations in the Tilebox Console](/docs/assets/console/storage-event-dark.png)

## Deleting Storage Event automations

To delete a registered storage event automation from Python, use `automations.delete`. You can also delete storage-event automations from the Tilebox Console. After deletion, no new jobs will be submitted by the storage event trigger. Past jobs already triggered will still remain queued.

```python title="Python"
from tilebox.workflows import Client

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

# delete the automation as returned by create_storage_event_automation
automations.delete(storage_event_automation)

# or manually by id:
automations.delete("0190bafc-b3b8-88c4-008b-a5db044380d0")
```

## Submitting Storage Event jobs manually

In Python, you can submit Storage Event tasks as regular tasks for testing purposes or as part of a larger workflow. To do so, instantiate the task with a specific storage location and object name using the `once` method.

```python title="Python"
job_client = client.jobs()

task = LogObjectCreation(head_bytes=20)

# submitting it directly won't work; raises ValueError:
# job_client.submit("manual-storage-event-job", task)

# instead, we specify a trigger condition, and submit a job manually
job_client.submit(
    "manual-storage-event-job",
    # simulate an event that occurred in the gcs bucket for the object "my-object.txt"
    task.once(gcs_bucket, "my-object.txt"),
)
```