> ## Documentation Index
> Fetch the complete documentation index at: /docs/llms.txt
> Use this file to discover all available pages before exploring further.

# triggers

# `prefect.server.events.triggers`

The triggers consumer watches events streaming in from the event bus and decides whether
to act on them based on the automations that users have set up.

## Functions

### `evaluate` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L89](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L89)</sup>

```python theme={null}
evaluate(session: AsyncSession, trigger: EventTrigger, bucket: 'ORMAutomationBucket', now: prefect.types._datetime.DateTime, triggering_event: Optional[ReceivedEvent]) -> 'ORMAutomationBucket | None'
```

Evaluates an Automation, either triggered by a specific event or proactively
on a time interval.  Evaluating a Automation updates the associated counters for
each automation, and will fire the associated action if it has met the threshold.

### `fire` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L330](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L330)</sup>

```python theme={null}
fire(session: AsyncSession, firing: Firing) -> None
```

### `evaluate_composite_trigger` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L341](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L341)</sup>

```python theme={null}
evaluate_composite_trigger(session: AsyncSession, firing: Firing) -> None
```

### `act` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L450](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L450)</sup>

```python theme={null}
act(firing: Firing) -> None
```

Given a Automation that has been triggered, the triggering labels and event
(if there was one), publish an action for the `actions` service to process.

### `update_events_clock` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L533](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L533)</sup>

```python theme={null}
update_events_clock(event: ReceivedEvent) -> None
```

### `get_events_clock` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L553](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L553)</sup>

```python theme={null}
get_events_clock() -> Optional[float]
```

### `get_events_clock_offset` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L558](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L558)</sup>

```python theme={null}
get_events_clock_offset() -> float
```

Calculate the current clock offset.  This takes into account both the `occurred`
of the last event, as well as the time we *saw* the last event.  This helps to
ensure that in low volume environments, we don't end up getting huge offsets.

### `reset_events_clock` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L574](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L574)</sup>

```python theme={null}
reset_events_clock() -> None
```

### `reactive_evaluation` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L581](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L581)</sup>

```python theme={null}
reactive_evaluation(event: ReceivedEvent, depth: int = 0) -> None
```

Evaluate all automations that may apply to this event.

event (ReceivedEvent): The event to evaluate. This object contains all the necessary information
about the event, including its type, associated resources, and metadata.
depth (int, optional): The current recursion depth. This is used to prevent infinite recursion
due to cyclic event dependencies. Defaults to 0 and is incremented with
each recursive call.

### `get_lost_followers` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L707](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L707)</sup>

```python theme={null}
get_lost_followers() -> List[ReceivedEvent]
```

Get followers that have been sitting around longer than our lookback

### `periodic_evaluation` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L712](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L712)</sup>

```python theme={null}
periodic_evaluation(now: prefect.types._datetime.DateTime) -> None
```

Periodic tasks that should be run regularly, but not as often as every event

### `evaluate_periodically` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L744](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L744)</sup>

```python theme={null}
evaluate_periodically(periodic_granularity: timedelta) -> None
```

Runs periodic evaluation on the given interval

### `find_interested_triggers` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L782](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L782)</sup>

```python theme={null}
find_interested_triggers(event: ReceivedEvent) -> Collection[EventTrigger]
```

### `clear_loaded_automations` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L791](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L791)</sup>

```python theme={null}
clear_loaded_automations() -> None
```

### `load_automation` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L819](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L819)</sup>

```python theme={null}
load_automation(automation: Optional[Automation]) -> None
```

Loads the given automation into memory so that it is available for evaluations

### `forget_automation` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L838](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L838)</sup>

```python theme={null}
forget_automation(automation_id: UUID) -> None
```

Unloads the given automation from memory

### `automation_changed` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L848](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L848)</sup>

```python theme={null}
automation_changed(automation_id: UUID, event: Literal['automation__created', 'automation__updated', 'automation__deleted']) -> None
```

### `load_automations` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L866](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L866)</sup>

```python theme={null}
load_automations(db: PrefectDBInterface, session: AsyncSession)
```

Loads all automations for the given set of accounts

### `read_automation_state_snapshot` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L882](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L882)</sup>

```python theme={null}
read_automation_state_snapshot(db: PrefectDBInterface, session: AsyncSession) -> AutomationStateSnapshot
```

### `reconcile_automations` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L900](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L900)</sup>

```python theme={null}
reconcile_automations(force: bool = False) -> bool
```

### `remove_buckets_exceeding_threshold` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L933](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L933)</sup>

```python theme={null}
remove_buckets_exceeding_threshold(db: PrefectDBInterface, session: AsyncSession, trigger: EventTrigger)
```

Deletes bucket where the count has already exceeded the threshold

### `read_buckets_for_automation` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L948](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L948)</sup>

```python theme={null}
read_buckets_for_automation(db: PrefectDBInterface, session: AsyncSession, trigger: Trigger, batch_size: int = AUTOMATION_BUCKET_BATCH_SIZE) -> AsyncGenerator['ORMAutomationBucket', None]
```

Yields buckets for the given automation and trigger in batches.

### `read_bucket` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L982](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L982)</sup>

```python theme={null}
read_bucket(db: PrefectDBInterface, session: AsyncSession, trigger: Trigger, bucketing_key: Tuple[str, ...]) -> Optional['ORMAutomationBucket']
```

Gets the bucket this event would fall into for the given Automation, if there is
one currently

### `read_bucket_by_trigger_id` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L999](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L999)</sup>

```python theme={null}
read_bucket_by_trigger_id(db: PrefectDBInterface, session: AsyncSession, automation_id: UUID, trigger_id: UUID, bucketing_key: Tuple[str, ...]) -> 'ORMAutomationBucket | None'
```

Gets the bucket this event would fall into for the given Automation, if there is
one currently

### `hold_expect_event` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1021](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1021)</sup>

```python theme={null}
hold_expect_event(session: AsyncSession, trigger: EventTrigger, bucketing_key: Tuple[str, ...], event: ReceivedEvent) -> 'ORMAutomationBucket | None'
```

Holds the event for every bucket that opens later over it, and returns this
key's current bucket for the caller to evaluate the event against.  Returns None
when the caller has nothing to evaluate: no bucket exists, or an earlier
evaluation of the same event held it.

### `increment_bucket` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1044](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1044)</sup>

```python theme={null}
increment_bucket(db: PrefectDBInterface, session: AsyncSession, bucket: 'ORMAutomationBucket', count: int, last_event: Optional[ReceivedEvent]) -> 'ORMAutomationBucket'
```

Adds the given count to the bucket, returning the new bucket

### `start_new_bucket` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1095](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1095)</sup>

```python theme={null}
start_new_bucket(db: PrefectDBInterface, session: AsyncSession, trigger: EventTrigger, bucketing_key: Tuple[str, ...], start: prefect.types._datetime.DateTime, end: prefect.types._datetime.DateTime, count: int, triggered_at: Optional[prefect.types._datetime.DateTime] = None, last_event: Optional[ReceivedEvent] = None) -> 'ORMAutomationBucket'
```

Ensures that a bucket with the given start and end exists with the given count
plus any held `expect` events that occurred inside its window, returning the new
bucket

### `ensure_bucket` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1164](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1164)</sup>

```python theme={null}
ensure_bucket(db: PrefectDBInterface, session: AsyncSession, trigger: EventTrigger, bucketing_key: Tuple[str, ...], start: prefect.types._datetime.DateTime, end: prefect.types._datetime.DateTime, last_event: Optional[ReceivedEvent], initial_count: int = 0) -> 'ORMAutomationBucket'
```

Ensures that a bucket has been started for the given automation and key,
returning the current bucket.  Will not modify the existing bucket.  A new bucket
starts at `initial_count` plus any held `expect` events that occurred inside its
window.

### `remove_bucket` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1228](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1228)</sup>

```python theme={null}
remove_bucket(db: PrefectDBInterface, session: AsyncSession, bucket: 'ORMAutomationBucket')
```

Removes the given bucket from the database

### `sweep_closed_buckets` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1242](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1242)</sup>

```python theme={null}
sweep_closed_buckets(db: PrefectDBInterface, session: AsyncSession, older_than: prefect.types._datetime.DateTime) -> None
```

### `reset` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1252](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1252)</sup>

```python theme={null}
reset() -> None
```

Resets the in-memory state of the service

### `listen_for_automation_changes` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1261](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1261)</sup>

```python theme={null}
listen_for_automation_changes() -> None
```

Listens for any changes to automations via PostgreSQL NOTIFY/LISTEN,
and applies those changes to the set of loaded automations.

### `consumer` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1332](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1332)</sup>

```python theme={null}
consumer(periodic_granularity: timedelta = timedelta(seconds=5)) -> AsyncGenerator[MessageHandler, None]
```

The `triggers.consumer` processes all Events arriving on the event bus to
determine if they meet the automation criteria, queuing up a corresponding
`TriggeredAction` for the `actions` service if the automation criteria is met.

### `proactive_evaluation` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1390](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1390)</sup>

```python theme={null}
proactive_evaluation(trigger: EventTrigger, as_of: prefect.types._datetime.DateTime) -> prefect.types._datetime.DateTime
```

The core proactive evaluation operation for a single Automation

### `evaluate_proactive_triggers` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1446](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/events/triggers.py#L1446)</sup>

```python theme={null}
evaluate_proactive_triggers() -> None
```