Add to Cursor Add to Claude Copy for LLM View as MD

Event Filters

Events can be filtered in Hatchet, which allows you to push events to Hatchet and only trigger task runs from them in certain cases. If you enable filters on a workflow, your workflow will be triggered once for each matching filter on any incoming event with a matching scope (more on scopes below).

Basic Usage

There are two ways to create filters in Hatchet.

Default filters on the workflow

The simplest way to create a filter is to register it declaratively with your workflow when it’s created. For example:

examples/python/events/worker.py

event_workflow_with_filter = hatchet.workflow(
    name="EventWorkflow",
    on_events=[EVENT_KEY, SECONDARY_KEY, WILDCARD_KEY],
    input_validator=EventWorkflowInput,
    default_filters=[
        DefaultFilter(
            expression="true",
            scope="example-scope",
            payload={
                "main_character": "Anna",
                "supporting_character": "Stiva",
                "location": "Moscow",
            },
        )
    ],
)

examples/typescript/on_event/workflow.ts

export const lowerWithFilter = hatchet.workflow<Input, LowerOutput>({
  name: 'lower',
  // 👀 Declare the event that will trigger the workflow
  onEvents: ['simple-event:create'],
  defaultFilters: [
    {
      expression: 'true',
      scope: 'example-scope',
      payload: {
        mainCharacter: 'Anna',
        supportingCharacter: 'Stiva',
        location: 'Moscow',
      },
    },
  ],
});

examples/go/on-event/main.go

func LowerWithFilter(client *hatchet.Client) *hatchet.StandaloneTask {
	return client.NewStandaloneTask(
		"lower", accessFilterPayload,
		hatchet.WithWorkflowEvents(SimpleEvent),
		hatchet.WithFilters(types.DefaultFilter{
			Expression: "true",
			Scope:      "example-scope",
			Payload: map[string]interface{}{
				"main_character":       "Anna",
				"supporting_character": "Stiva",
				"location":             "Moscow"},
		}),
	)
}

The Ruby SDK is in early access, and may change. We'd love your feedback!

examples/ruby/events/worker.rb

EVENT_WORKFLOW_WITH_FILTER = HATCHET.workflow(
  name: "EventWorkflow",
  on_events: [EVENT_KEY, SECONDARY_KEY, WILDCARD_KEY],
  default_filters: [
    Hatchet::DefaultFilter.new(
      expression: "true",
      scope: "example-scope",
      payload: {
        "main_character" => "Anna",
        "supporting_character" => "Stiva",
        "location" => "Moscow"
      }
    )
  ]
)

EVENT_WORKFLOW.task(:task) do |input, ctx|
  puts "event received"
  ctx.filter_payload
end

In each of these cases, we register a filter with the workflow. Note that these “declarative” filters are overwritten each time your workflow is updated, so the ids associated with them will not be stable over time. This allows you to modify a filter in-place or remove a filter, and not need to manually delete it over the API.

Filters feature client

You also can create event filters by using the filters clients on the SDKs:

examples/python/events/filter.py

hatchet.filters.create(
    workflow_id=event_workflow.id,
    expression="input.should_skip == false",
    scope="foobarbaz",
    payload={
        "main_character": "Anna",
        "supporting_character": "Stiva",
        "location": "Moscow",
    },
)

⚠️

Note the scope argument to the filter. When you create a filter, it must be given a scope which will be used by Hatchet internally to look it up. When you push events that you want filtered, you must provide a scope with those events that matches the scope sent with the filter. If you do not, the filter will not apply.

Then, push an event that uses the filter to determine whether or not to run. For instance, this run will be skipped, since the payload does not match the expression:

examples/python/events/filter.py

hatchet.event.push(
    event_key=EVENT_KEY,
    payload={
        "should_skip": True,
    },
    options=PushEventOptions(
        scope="foobarbaz",
    ),
)

But this one will be triggered since the payload does match the expression:

examples/python/events/filter.py

hatchet.event.push(
    event_key=EVENT_KEY,
    payload={
        "should_skip": False,
    },
    options=PushEventOptions(
        scope="foobarbaz",
    ),
)

In Hatchet, filters are “positive”, meaning that we look for matches to the filter to determine which tasks to trigger.

Accessing the filter payload

You can access the filter payload by using the Context in the task that was triggered by your event:

examples/python/events/worker.py

@event_workflow_with_filter.task()
def filtered_task(input: EventWorkflowInput, ctx: Context) -> None:
    print(ctx.filter_payload)

Advanced Usage

In addition to referencing input in the expression (which corresponds to the event payload), you can also reference the following fields:

payload corresponds to the filter payload (which was part of the request when the filter was created).
additional_metadata allows for filtering based on additional_metadata sent with the event.
event_key allows for filtering based on the key of the event, such as user:created.

Event Trigger Inter-Service Triggering

We use cookies

Event Filters

Basic Usage

Default filters on the workflow

Filters feature client

Accessing the filter payload

Advanced Usage