---
question: How do I start a job when a file lands or a key changes?
kind: task
via_web: na
via_cli: na
via_python: available
via_api: na
via_config: na
ref_python: OnEvent
prereq: You need a NATS project (the default of `abi new project`) and a module. [Create a module](/capabilities/build/modules/creating-a-module) first.
next:
  - /capabilities/build/orchestrations/chain-jobs-with-events
  - /capabilities/build/platform/events-overview
---

# React to an event

`OnEvent` starts a job when something happens: an ABI event of a given type, such as an object written to storage, or any message on a NATS subject. Each replica of the module listens; one of them turns each matching event into a run.

## Web UI

Not applicable. No screen declares an event trigger: it is code. Nexus lists the trigger under System › Jobs, but shows an ABI event trigger as `On event evt.1a2b…`, the start of a hash of the event type: not the type's name, and not its filter. Look for the job by name, and keep the type and filter in the first line of its docstring, which Nexus shows as the job's description.

## CLI

Not applicable. No `abi` command declares a trigger. Use Python.

## Python

### Listen to an event type

Pass the event type's IRI and, usually, a filter. This job runs when a `.csv` file is written under `reports/` in object storage:

```python title="src/reports/__init__.py"
from naas_abi_sdk import BaseModule, ModuleDependencies
from naas_abi_sdk.jobs import JobContext, OnEvent, job

OBJECT_PUT = "http://ontology.naas.ai/abi/object_storage/ObjectPut"


class ABIModule(BaseModule):
    dependencies = ModuleDependencies(services=("object_storage",))

    @job(triggers=(
        OnEvent(event_type=OBJECT_PUT,
                filter={"prefix": {"prefix": "reports/"}, "key": {"suffix": ".csv"}}),
    ))
    async def load_report(self, ctx: JobContext) -> dict:
        """Load each CSV report written under reports/."""
        put = ctx.payload["data"]  # the event's fields
        content = await self.engine.services.object_storage.get_object(put["prefix"], put["key"])
        rows = await self.load_rows(content)  # your code
        return {"key": put["key"], "rows": rows}
```

`ctx.payload` holds `subject` (the NATS subject the event came on) and `data` (the event as JSON: its fields, plus `_class_uri`, `_uri` and `created_at`).

The filter is checked before a run is created: an event it rejects leaves no trace under System › Jobs.

### Or listen to a NATS subject

Pass a subject instead of an event type, with NATS wildcards (`*` for one token, `>` for the rest). The job runs on every message published there, by any client:

```python
OnEvent("exports.finished.>")
```

A message that is JSON arrives decoded in `ctx.payload["data"]`; any other body arrives as text. A filter works the same way on a JSON body.

`OnEvent` takes the subject or `event_type`, never both.

### Write the filter

A filter is a dict. Each key is a dotted path into the event's JSON (`key`, `actor.id`); its value is what that field must match. **Every key must match** for the event to start the job.

| Value | Matches when the field |
|---|---|
| `"reports/daily"` (a scalar) | equals it, compared as text |
| `["daily", "weekly"]` (a list) | is one of them |
| `{"eq": "daily"}` | equals it, compared as text |
| `{"ne": "draft"}` | is set and differs from it |
| `{"in": ["daily", "weekly"]}` | is one of them |
| `{"gt": 0}`, `{"gte": 0}`, `{"lt": 1048576}`, `{"lte": 1048576}` | is a number above or below it |
| `{"prefix": "reports/"}` | is text that starts with it |
| `{"suffix": ".csv"}` | is text that ends with it |
| `{"contains": "daily"}` | is text that contains it |
| `{"exists": True}`, `{"exists": False}` | is set, or missing (or `null`) |

Several operators in one dict must all match: `{"size_bytes": {"gt": 0, "lte": 1048576}}`. An unknown operator, or a key that is not a dotted path, raises `ValueError` when the module class loads.

### Start the module

The trigger is live once the module has started (restart the engine for an in-engine module). Write a matching object, then open System › Jobs: the job has a new run, tagged **Event**.

## REST API

Not applicable. No route declares a trigger. To start the job without an event, use `POST /api/admin/system/jobs/<module>/<job>/trigger`: see [Run a job by hand](/capabilities/build/orchestrations/run-a-job-by-hand?via=api).

## Config

Not applicable. `config.yaml` has no per-job key: triggers live in code.

## Kernel events

The engine's services publish these events as they work, when the engine runs the event service (a new project does). The IRI is what `event_type` takes. Every event also carries `created_at`, and, when the request had them, `actor_user_id`, `actor_workspace_id` and `triggered_via`.

| Service | Event type IRI | Fields worth filtering on |
|---|---|---|
| Object storage | `http://ontology.naas.ai/abi/object_storage/ObjectPut` | `prefix`, `key`, `size_bytes` |
| Object storage | `http://ontology.naas.ai/abi/object_storage/ObjectDeleted` | `prefix`, `key` |
| Key-value | `http://ontology.naas.ai/abi/keyvalue/KeyValueSet` | `key`, `size_bytes`, `ttl_seconds` |
| Key-value | `http://ontology.naas.ai/abi/keyvalue/KeyValueDeleted` | `key` |
| Key-value | `http://ontology.naas.ai/abi/keyvalue/KeyValueError` | `key`, `operation`, `message` |
| Triple store | `http://ontology.naas.ai/abi/triple_store/TriplesInserted` | `graph_name`, `triple_count` |
| Triple store | `http://ontology.naas.ai/abi/triple_store/TriplesRemoved` | `graph_name`, `triple_count` |
| Triple store | `http://ontology.naas.ai/abi/triple_store/GraphCreated` | `graph_name` |
| Triple store | `http://ontology.naas.ai/abi/triple_store/GraphCleared` | `graph_name` |
| Triple store | `http://ontology.naas.ai/abi/triple_store/GraphDropped` | `graph_name` |
| Triple store | `http://ontology.naas.ai/abi/triple_store/SchemaLoaded` | `filepath` |
| Triple store | `http://ontology.naas.ai/abi/triple_store/SchemaRemoved` | `filepath` |
| Triple store | `http://ontology.naas.ai/abi/triple_store/TripleStoreError` | `graph_name`, `filepath`, `operation`, `message` |
| Cache | `http://ontology.naas.ai/abi/cache/CacheSet` | `key`, `tier`, `data_type`, `size_bytes` |
| Cache | `http://ontology.naas.ai/abi/cache/CacheDeleted` | `key`, `tier` |
| Cache | `http://ontology.naas.ai/abi/cache/CacheError` | `key`, `tier`, `operation`, `message` |
| Vector store | `http://ontology.naas.ai/abi/vector_store/CollectionEnsured` | `collection_name` |
| Vector store | `http://ontology.naas.ai/abi/vector_store/DocumentsAdded` | `collection_name`, `document_count` |
| Vector store | `http://ontology.naas.ai/abi/vector_store/DocumentUpdated` | `collection_name`, `document_id` |
| Vector store | `http://ontology.naas.ai/abi/vector_store/DocumentsDeleted` | `collection_name`, `document_count` |
| Vector store | `http://ontology.naas.ai/abi/vector_store/CollectionDeleted` | `collection_name` |
| Vector store | `http://ontology.naas.ai/abi/vector_store/VectorStoreError` | `collection_name`, `operation`, `message` |
| Email | `http://ontology.naas.ai/abi/email/EmailSent` | `to`, `subject`, `sender` |
| Email | `http://ontology.naas.ai/abi/email/EmailError` | `to`, `subject`, `message` |
| Secrets | `http://ontology.naas.ai/abi/secret/SecretSet` | `key` (never the value) |
| Secrets | `http://ontology.naas.ai/abi/secret/SecretRemoved` | `key` |
| Secrets | `http://ontology.naas.ai/abi/secret/SecretError` | `key`, `operation`, `message` |
| Datasets | `http://ontology.naas.ai/abi/dataset/DatasetCatalogPressure` | `dataset_name`, `namespace`, `inlined_records`, `threshold_records` |

`DatasetCatalogPressure` is a warning: a dataset has at least `threshold_records` rows waiting for compaction. Your own modules can publish events of their own types: see [Chain jobs with events](/capabilities/build/orchestrations/chain-jobs-with-events#publish-an-event-of-your-own).

## Delivery is at most once

An event reaches a job only if a replica of the module is listening when it is published. An event published while the module is stopped, restarting or not yet started is lost to the trigger. Two things follow:

- **Do not count on the trigger alone** when every event matters. Let the event wake the job, and have the job read what it has not handled from the event log, which keeps every event. That pattern is [Wake up, then read the log](/capabilities/build/orchestrations/chain-jobs-with-events#wake-up-then-read-the-log).
- **Expect repeats.** Once an event became a trigger, the run is at least once, like any job run: it can start again after a crash. Make the job safe to run twice.

A run waits up to 24 hours for a free slot once its event was received; past that, the trigger expires.