---
sidebar:
  hidden: true
title: roboto.domain.actions.trigger
---
## Module Contents

### Trigger

```python
class roboto.domain.actions.trigger.Trigger(
    record: roboto.domain.actions.trigger_record.TriggerRecord,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
)
```

`from roboto import Trigger`

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/trigger.py#L57-L540)

A rule that automatically invokes an action when specific events or conditions occur.

> **Deprecated since version This**
>
> is the legacy trigger model, kept so existing code keeps working. It cannot express triggers made with the current model (several targets, agent or Slack targets, platform events beyond files and datasets) and will be removed. Use [`roboto.domain.triggers.Trigger`](/reference/python-sdk/roboto/domain/triggers/trigger#roboto.domain.triggers.trigger.Trigger), which reads and edits every trigger.

Triggers enable automated data processing workflows by monitoring for specific events (like new datasets being created) and automatically invoking actions when conditions are met. They eliminate the need for manual intervention in routine data processing tasks.

Triggers can be configured to:

- Monitor for new datasets, files, or other data sources
- Apply conditional logic to determine when to execute
- Specify input data patterns and action parameters
- Override compute requirements and container parameters
- Execute actions for each matching item or in batch

A trigger consists of:

- Target action to invoke
- Input data requirements and patterns
- Execution conditions and causes
- Parameter values and overrides
- Scheduling and execution settings

**Parameters**

- **record** (`roboto.domain.actions.trigger_record.TriggerRecord`)
- **roboto_client** (`Optional[roboto.http.RobotoClient]`)

**Properties**

- **Trigger.condition** (`roboto.query.ConditionType | None`): Return type: `Optional[roboto.query.ConditionType]`
- **Trigger.created** (`datetime.datetime`)
- **Trigger.created_by** (`str`)
- **Trigger.enabled** (`bool`)
- **Trigger.for_each** (`roboto.domain.actions.trigger_record.TriggerForEachPrimitive`)
- **Trigger.modified** (`datetime.datetime`)
- **Trigger.modified_by** (`str`)
- **Trigger.name**
- **Trigger.org_id**
- **Trigger.record** (`roboto.domain.actions.trigger_record.TriggerRecord`)
- **Trigger.service_user_id** (`str | None`): Return type: `Optional[str]`
- **Trigger.trigger_id** (`str`)

#### Trigger.create()

```python
@classmethod
def create(
    name: str,
    action_name: str,
    required_inputs: list[str],
    for_each: roboto.domain.actions.trigger_record.TriggerForEachPrimitive,
    enabled: bool = True,
    action_digest: Optional[str] = None,
    action_owner_id: Optional[str] = None,
    additional_inputs: Optional[list[str]] = None,
    causes: Optional[list[roboto.domain.actions.trigger_record.TriggerEvaluationCause]] = None,
    compute_requirement_overrides: Optional[roboto.domain.actions.invocation_record.ComputeRequirements] = None,
    condition: Optional[roboto.query.ConditionType] = None,
    container_parameter_overrides: Optional[roboto.domain.actions.invocation_record.ContainerParameters] = None,
    parameter_values: Optional[dict[str, Any]] = None,
    service_user_id: Optional[str] = None,
    timeout: Optional[int] = None,
    caller_org_id: Optional[str] = None,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
) -> Trigger
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/trigger.py#L149-L272)

Create a new trigger that automatically invokes an action when conditions are met.

Creates a trigger that monitors for specific events (like new datasets or files) and automatically invokes the specified action when the trigger conditions are satisfied. This enables automated data processing workflows.

**Parameters**

- **name** (`str`): Unique name for the trigger within the organization.
- **action_name** (`str`): Name of the action to invoke when the trigger fires.
- **required_inputs** (`list[str]`): List of file patterns that must be present for the trigger to fire. Uses glob patterns like "\*\*/\*.bag" or "data/\*.csv".
- **for_each** (`roboto.domain.actions.trigger_record.TriggerForEachPrimitive`): Granularity of execution - Dataset creates one invocation per dataset, DatasetFile creates one invocation per matching file.
- **enabled** (`bool`): Whether the trigger should be active immediately after creation.
- **action_digest** (`Optional[str]`): Specific version digest of the action to invoke. If not provided, uses the latest version.
- **action_owner_id** (`Optional[str]`): Organization ID that owns the target action. If not provided, searches in the caller's organization.
- **additional_inputs** (`Optional[list[str]]`): Optional additional file patterns to include in invocations.
- **causes** (`Optional[list[roboto.domain.actions.trigger_record.TriggerEvaluationCause]]`): List of events that can cause this trigger to be evaluated. If not provided, uses default causes.
- **compute_requirement_overrides** (`Optional[roboto.domain.actions.invocation_record.ComputeRequirements]`): Optional compute requirement overrides for action invocations.
- **condition** (`Optional[roboto.query.ConditionType]`): Optional condition that must be met for the trigger to fire. Can filter based on metadata, file properties, etc.
- **container_parameter_overrides** (`Optional[roboto.domain.actions.invocation_record.ContainerParameters]`): Optional container parameter overrides for action invocations.
- **parameter_values** (`Optional[dict[str, Any]]`): Parameter values to pass to the action when invoked.
- **service_user_id** (`Optional[str]`): Optional service user ID for authentication.
- **timeout** (`Optional[int]`): Optional timeout override for action invocations in minutes.
- **caller_org_id** (`Optional[str]`): Organization ID to create the trigger in. Defaults to caller's org.
- **roboto_client** (`Optional[roboto.http.RobotoClient]`): Roboto client instance. Uses default if not provided.

**Returns**

- `Trigger`: The newly created Trigger instance.

**Raises**

- [`RobotoIllegalArgumentException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoIllegalArgumentException): If the trigger configuration is invalid.
- [`RobotoInvalidRequestException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoInvalidRequestException): If the request is malformed.
- [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): If the caller lacks permission to create triggers.

**Usage**

Create a simple trigger for ROS bag files:

```python
from roboto.domain.actions import Trigger, TriggerForEachPrimitive
trigger = Trigger.create(
    name="auto_process_bags",
    action_name="ros_ingestion",
    required_inputs=["**/*.bag"],
    for_each=TriggerForEachPrimitive.Dataset,
)
```

Create a conditional trigger with parameters:

```python
from roboto.query import Condition
condition = Condition("metadata.sensor_type").equals("lidar")
trigger = Trigger.create(
    name="lidar_processing",
    action_name="lidar_processor",
    required_inputs=["**/*.pcd"],
    for_each=TriggerForEachPrimitive.Dataset,
    condition=condition,
    parameter_values={"resolution": "high", "filter": "statistical"},
)
```

Create a trigger with compute overrides:

```python
from roboto.domain.actions import ComputeRequirements
trigger = Trigger.create(
    name="heavy_processing",
    action_name="ml_inference",
    required_inputs=["**/*.jpg", "**/*.png"],
    for_each=TriggerForEachPrimitive.DatasetFile,
    compute_requirement_overrides=ComputeRequirements(vCPU=8192, memory=16384),
)
```

#### Trigger.delete()

```python
def delete()
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/trigger.py#L381-L386)

#### Trigger.disable()

```python
def disable()
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/trigger.py#L388-L389)

#### Trigger.enable()

```python
def enable()
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/trigger.py#L391-L392)

#### Trigger.from_name()

```python
@classmethod
def from_name(
    name: str,
    owner_org_id: Optional[str] = None,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
) -> Trigger
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/trigger.py#L275-L288)

**Parameters**

- **name** (`str`)
- **owner_org_id** (`Optional[str]`)
- **roboto_client** (`Optional[roboto.http.RobotoClient]`)

**Returns**

- `Trigger`

#### Trigger.get_action()

```python
def get_action() -> roboto.domain.actions.action.Action
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/trigger.py#L394-L400)

**Returns**

- `roboto.domain.actions.action.Action`

#### Trigger.get_evaluations()

```python
def get_evaluations(
    limit: Optional[int] = None,
    page_token: Optional[str] = None,
) -> collections.abc.Generator[roboto.domain.actions.trigger_record.TriggerEvaluationRecord, None, None]
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/trigger.py#L402-L426)

**Parameters**

- **limit** (`Optional[int]`)
- **page_token** (`Optional[str]`)

**Returns**

- `collections.abc.Generator[roboto.domain.actions.trigger_record.TriggerEvaluationRecord, None, None]`

#### Trigger.get_evaluations_for_dataset()

```python
@staticmethod
def get_evaluations_for_dataset(
    dataset_id: str,
    owner_org_id: Optional[str] = None,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
) -> collections.abc.Generator[roboto.domain.actions.trigger_record.TriggerEvaluationRecord, None, None]
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/trigger.py#L92-L146)

Get all trigger evaluations for a specific dataset.

Retrieves the history of trigger evaluations that were performed for a given dataset, including successful invocations and failed attempts.

**Parameters**

- **dataset_id** (`str`): The ID of the dataset to get evaluations for.
- **owner_org_id** (`Optional[str]`): Organization ID that owns the dataset. If not provided, searches in the caller's organization.
- **roboto_client** (`Optional[roboto.http.RobotoClient]`): Roboto client instance. Uses default if not provided.

**Yields**

- TriggerEvaluationRecord instances for the dataset.

**Raises**

- [`RobotoNotFoundException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoNotFoundException): If the dataset is not found.
- [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): If the caller lacks permission to access evaluations.

**Returns**

- `collections.abc.Generator[roboto.domain.actions.trigger_record.TriggerEvaluationRecord, None, None]`

**Usage**

Get all evaluations for a dataset:

```python
for evaluation in Trigger.get_evaluations_for_dataset("ds_12345"):
    print(f"Trigger: {evaluation.trigger_name}, Status: {evaluation.status}")
```

Check if any triggers succeeded for a dataset:

```python
from roboto.domain.actions import TriggerEvaluationStatus
evaluations = list(Trigger.get_evaluations_for_dataset("ds_12345"))
successful = [e for e in evaluations if e.status == TriggerEvaluationStatus.Succeeded]
print(f"Found {len(successful)} successful trigger evaluations")
```

#### Trigger.get_invocations()

```python
def get_invocations() -> collections.abc.Generator[roboto.domain.actions.invocation.Invocation, None, None]
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/trigger.py#L428-L440)

**Returns**

- `collections.abc.Generator[roboto.domain.actions.invocation.Invocation, None, None]`

#### Trigger.invoke()

```python
def invoke(
    data_source: roboto.domain.actions.invocation_record.InvocationDataSource,
    idempotency_id: Optional[str] = None,
    input_data_override: Optional[list[str]] = None,
    upload_destination: Optional[roboto.domain.actions.invocation_record.InvocationUploadDestination] = None,
) -> Optional[roboto.domain.actions.invocation.Invocation]
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/trigger.py#L446-L470)

**Parameters**

- **data_source** (`roboto.domain.actions.invocation_record.InvocationDataSource`)
- **idempotency_id** (`Optional[str]`)
- **input_data_override** (`Optional[list[str]]`)
- **upload_destination** (`Optional[roboto.domain.actions.invocation_record.InvocationUploadDestination]`)

**Returns**

- `Optional[roboto.domain.actions.invocation.Invocation]`

#### Trigger.latest_evaluation()

```python
def latest_evaluation() -> Optional[roboto.domain.actions.trigger_record.TriggerEvaluationRecord]
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/trigger.py#L442-L444)

**Returns**

- `Optional[roboto.domain.actions.trigger_record.TriggerEvaluationRecord]`

#### Trigger.query()

```python
@classmethod
def query(
    spec: Optional[roboto.query.QuerySpecification] = None,
    owner_org_id: Optional[str] = None,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
) -> collections.abc.Generator[Trigger, None, None]
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/trigger.py#L291-L314)

**Parameters**

- **spec** (`Optional[roboto.query.QuerySpecification]`)
- **owner_org_id** (`Optional[str]`)
- **roboto_client** (`Optional[roboto.http.RobotoClient]`)

**Returns**

- `collections.abc.Generator[Trigger, None, None]`

#### Trigger.to_dict()

```python
def to_dict() -> dict[str, Any]
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/trigger.py#L472-L473)

**Returns**

- `dict[str, Any]`

#### Trigger.update()

```python
def update(
    action_name: Union[str, roboto.sentinels.NotSetType] = NotSet,
    action_owner_id: Union[str, roboto.sentinels.NotSetType] = NotSet,
    action_digest: Optional[Union[str, roboto.sentinels.NotSetType]] = NotSet,
    additional_inputs: Optional[Union[list[str], roboto.sentinels.NotSetType]] = NotSet,
    causes: Union[list[roboto.domain.actions.trigger_record.TriggerEvaluationCause], roboto.sentinels.NotSetType] = NotSet,
    compute_requirement_overrides: Optional[Union[roboto.domain.actions.invocation_record.ComputeRequirements, roboto.sentinels.NotSetType]] = NotSet,
    container_parameter_overrides: Optional[Union[roboto.domain.actions.invocation_record.ContainerParameters, roboto.sentinels.NotSetType]] = NotSet,
    condition: Optional[Union[roboto.query.ConditionType, roboto.sentinels.NotSetType]] = NotSet,
    enabled: Union[bool, roboto.sentinels.NotSetType] = NotSet,
    for_each: Union[roboto.domain.actions.trigger_record.TriggerForEachPrimitive, roboto.sentinels.NotSetType] = NotSet,
    parameter_values: Optional[Union[dict[str, Any], roboto.sentinels.NotSetType]] = NotSet,
    required_inputs: Union[list[str], roboto.sentinels.NotSetType] = NotSet,
    timeout: Optional[Union[int, roboto.sentinels.NotSetType]] = NotSet,
) -> Trigger
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/trigger.py#L475-L517)

**Parameters**

- **action_name** (`Union[str, roboto.sentinels.NotSetType]`)
- **action_owner_id** (`Union[str, roboto.sentinels.NotSetType]`)
- **action_digest** (`Optional[Union[str, roboto.sentinels.NotSetType]]`)
- **additional_inputs** (`Optional[Union[list[str], roboto.sentinels.NotSetType]]`)
- **causes** (`Union[list[roboto.domain.actions.trigger_record.TriggerEvaluationCause], roboto.sentinels.NotSetType]`)
- **compute_requirement_overrides** (`Optional[Union[roboto.domain.actions.invocation_record.ComputeRequirements, roboto.sentinels.NotSetType]]`)
- **container_parameter_overrides** (`Optional[Union[roboto.domain.actions.invocation_record.ContainerParameters, roboto.sentinels.NotSetType]]`)
- **condition** (`Optional[Union[roboto.query.ConditionType, roboto.sentinels.NotSetType]]`)
- **enabled** (`Union[bool, roboto.sentinels.NotSetType]`)
- **for_each** (`Union[roboto.domain.actions.trigger_record.TriggerForEachPrimitive, roboto.sentinels.NotSetType]`)
- **parameter_values** (`Optional[Union[dict[str, Any], roboto.sentinels.NotSetType]]`)
- **required_inputs** (`Union[list[str], roboto.sentinels.NotSetType]`)
- **timeout** (`Optional[Union[int, roboto.sentinels.NotSetType]]`)

**Returns**

- `Trigger`

#### Trigger.wait_for_evaluations_to_complete()

```python
def wait_for_evaluations_to_complete(
    timeout: float = 60 * 5,
    poll_interval: roboto.waiters.Interval = 5,
) -> None
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/trigger.py#L519-L540)

Wait for all evaluations for this trigger to complete.

Throws a [`TimeoutError`](/reference/python-sdk/roboto/waiters#roboto.waiters.TimeoutError) if the timeout is reached.

**Parameters**

- **timeout** (`float`): The maximum amount of time, in seconds, to wait for the evaluations to complete.
- **poll_interval** (`roboto.waiters.Interval`): The amount of time, in seconds, to wait between polling iterations.

**Returns**

- `None`