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

# Triggers for data workflows

> Automate workflow execution with schedule-based, data modeling, or records triggers.

A data workflow can have one or more triggers associated with it, each uniquely identified by an external ID.

<Tip>
  See also [API workflow triggers](/api-reference/concepts/20230101/workflow-triggers).
</Tip>

## Pass data to workflows

When a trigger starts, it creates a workflow execution and passes data through the workflow input. This data becomes accessible to all tasks in the workflow through [dynamic references](/cdf/data_workflows/task_types#dynamic-references) using the `${workflow.input}` syntax. The workflow input can contain a combination of:

* **Static input**: Data you define when creating the trigger.

* **Dynamic input**: Data collected by the trigger. For data modeling and records triggers, matching items are available as `${workflow.input.items}`.

* **System input**: Metadata added by the system, for instance `${workflow.input.version}`.

When you configure **Transformation**, **Function**, or **Agent** tasks, the [reference picker](/cdf/data_workflows/task_types#reference-picker-in-the-editor) suggests paths from your trigger's **static input** and other inferable workflow input fields. For deeper runtime shapes (for example inside `${workflow.input.items}`), add **custom** segments as described in [Known and custom path segments](/cdf/data_workflows/task_types#known-and-custom-path-segments). Other task types use `${workflow.input...}` in definitions or the API.

## Set up trigger rules

Specify a `triggerRule` to define the conditions for running a trigger. `triggerRule` consists of a trigger type and its associated parameters.

### Schedule a data workflow

Use the `schedule` trigger type to run a data workflow at regular intervals. The interval is specified by a [cron expression](https://en.wikipedia.org/wiki/Cron).

For example, to run the trigger every day at 12:00 AM, use the cron expression `"0 0 * * *"`. The time zone for the cron expression is UTC.

Scheduled triggers pass only the static input data you define when you create a trigger, and system metadata, such as workflow version.

### Trigger on data modeling events

Use the `dataModeling` trigger type to run a data workflow based on changes to data modeling instances matching a filter.

Specify the filter using the trigger's `dataModelingQuery` parameter. See also [data modeling queries](/cdf/dm/dm_concepts/dm_querying#how-to-define-a-graph-query).

Data modeling triggers use a **change-based polling mechanism**:

* **Polling**: The system periodically checks for instances matching your filter criteria.
* **Change detection**: Triggers detect changes based on the `lastUpdatedTime` of instances.
* **Batching**: Multiple matching instances are collected into batches.
* **Execution**: When batch criteria are met, a workflow execution starts with the collected instances as input.

#### Prevent excessive data modeling trigger runs

Data modeling triggers fire on any change that updates the `lastUpdatedTime` property of instances matching your filter. To prevent excessive trigger runs:

* **Use specific filters**: Ensure your filter is specific enough to avoid matching instances that shouldn't trigger the workflow.
* **Implement processing flags**: Include a property like `isProcessed` in your data model to track workflow completion.
  * Filter for instances where `isProcessed` is `False`.
  * Update `isProcessed` to `True` after processing to prevent re-triggering on the same instance

#### Control batching

You can control the batching of changes using the following parameters:

* `batchSize`: The maximum number of items to pass as input to a workflow execution.
* `batchTimeout`: The maximum time in seconds to wait for the batch to complete before passing it to a workflow execution.

After the `batchTimeout` has passed, a partial batch will be passed to the workflow execution. A complete batch will be passed to the workflow execution without further delay.

#### Match `batchSize` with data modeling query limits

When using a `dataModelingQuery`, each `with` clause that is referenced in the `select` block **must specify a `limit` equal to the trigger’s `batchSize`**.

This ensures that the number of instances returned by the data modeling query matches the size of the workflow input batch.

If the `limit` is missing or does not match `batchSize`, the trigger will fail validation.

* `with` clauses **referenced** in `select`:\
  Must include a `limit` equal to `batchSize`.

* `with` clauses **not referenced** in `select`:\
  Can omit or use a different `limit`; they are validated separately by the data modeling API.

There is **no default `limit`** — it must always be explicitly defined in the query.

#### Example: data modeling trigger

This example shows how to create a trigger using the Python SDK. The trigger starts a workflow based on changes to instances of `ExampleView` with the property `site` equal to *Site A*. According to the batch configurations, the trigger starts the workflow when it reaches 100 instance changes or 60 seconds after the first change.

```python theme={"languages":{"custom":["/_languages/kuiper.json","../_languages/kuiper.json"]}}
# Specify the relevant data model view
example_view_id = ViewId(space="ExampleSpace", external_id="ExampleView", version="1")

# Define the trigger rule, including the data modeling query
trigger_rule = WorkflowDataModelingTriggerRule(
    data_modeling_query=WorkflowTriggerDataModelingQuery(
        # Query instances of ExampleView with property site="Site A"
        with_={
            "example_view": NodeResultSetExpression(
                limit=100, # a with clause referenced by a select should always equal batchSize
                filter=Equals(
                    property=example_view_id.as_property_ref("site"), value="Site A"
                )
            )
        },
        # Return properties "site" and "name" from the instances
        select={
            "example_view": Select(
                sources=[
                    SourceSelector(source=example_view_id, properties=["site", "name"])
                ]
            )
        },
    ),
    batch_size=100,  # Start the workflow when reaching 100 changes
    batch_timeout=60,  # Start the workflow when reaching 60 seconds after the first change
)

# Create the trigger
client.workflows.triggers.upsert(
   WorkflowTriggerUpsert(
    external_id="my-data-modeling-trigger",
    trigger_rule=trigger_rule,
    workflow_external_id="my-workflow-external-id",
    workflow_version="v1",
   )
)
```

### Trigger on record changes

<Warning>
  The features described in this section are in [public preview](/cdf/product_feature_status#public-preview) and may change.
</Warning>

Use the `recordStream` trigger type to run a data workflow when [records](/cdf/dm/records/concepts/records_and_streams) change in a stream. Specify the stream using the trigger's `streamExternalId` parameter.

You can optionally provide a `filter` to limit which records in the stream trigger the workflow, and a `sources` list to select which container properties to include in the workflow input. If you omit `sources`, all properties from the record's containers are included.

Records triggers use a **change-based batching mechanism** similar to data modeling triggers:

* **Polling**: The system periodically checks for record changes in the specified stream.
* **Filtering**: If a filter is defined, only matching records trigger the workflow.
* **Batching**: Matching records are collected into batches.
* **Execution**: When batch criteria are met, a workflow execution starts with the collected records as input.

#### Control batching

You can control the batching of changes using the following parameters:

* `batchSize`: The maximum number of records to pass as input to a workflow execution. Range: 1-1000.
* `batchTimeout`: The maximum time in seconds to wait for the batch to fill before passing it to a workflow execution. Range: 10-86400.

After the `batchTimeout` has passed, a partial batch is passed to the workflow execution. A complete batch is passed without further delay.

## Target triggers

The trigger targets a data workflow, defined by a `workflowExternalId` and `workflowVersion`.

## Trigger input data

You can define an `input` data object as part of the trigger. When the trigger starts executing the target data workflow, this input object will be provided as input to the workflow.

### Static input data

Static input data is defined when creating the trigger and remains constant for all executions:

```json theme={"languages":{"custom":["/_languages/kuiper.json","../_languages/kuiper.json"]}}
{
  "external_id": "my-trigger",
  "input": {
    "environment": "production",
    "notification_email": "admin@company.com"
  }
}
```

This data is accessible in workflow tasks as `${workflow.input.environment}` and `${workflow.input.notification_email}`.

### Dynamic input: Data modeling triggers

For data modeling triggers, the system automatically adds an `items` array to the workflow input containing the instances that matched the trigger criteria:

```json theme={"languages":{"custom":["/_languages/kuiper.json","../_languages/kuiper.json"]}}
{
  "version": "v1",
  "items": [
    {
      "instanceType": "node",
      "externalId": "instance1",
      "space": "mySpace",
      "properties": {
        "mySpace": {
          "myView/version": {
            "site": "Site A",
            "name": "Equipment 1",
            "email": "user@company.com"
          }
        }
      }
    },
    {
      "instanceType": "node",
      "externalId": "instance2",
      "space": "mySpace",
      "properties": {
        "mySpace": {
          "myView/version": {
            "site": "Site A",
            "name": "Equipment 2",
            "email": "admin@company.com"
          }
        }
      }
    }
  ]
}
```

Access this data in your tasks using `${workflow.input.items}`. Each item in the array represents a data modeling instance that matched the trigger criteria, including the properties you selected in the data modeling query.

### Dynamic input: Records triggers

For records triggers, the system automatically adds an `items` array to the workflow input containing the records that matched the trigger criteria:

```json theme={"languages":{"custom":["/_languages/kuiper.json","../_languages/kuiper.json"]}}
{
  "version": "v1",
  "items": [
    {
      "externalId": "alarm-001",
      "space": "mySpace",
      "properties": {
        "mySpace": {
          "alarms": {
            "severity": "high",
            "source": "compressor-5",
            "message": "Temperature threshold exceeded"
          }
        }
      }
    },
    {
      "externalId": "alarm-002",
      "space": "mySpace",
      "properties": {
        "mySpace": {
          "alarms": {
            "severity": "medium",
            "source": "pump-3",
            "message": "Vibration anomaly detected"
          }
        }
      }
    }
  ]
}
```

Access this data in your tasks using `${workflow.input.items}`. Each item in the array represents a record from the stream that matched the trigger criteria. If you specified `sources` in the trigger configuration, only the selected container properties are included. Otherwise, all properties from the record's containers are included.

### Process multiple items

Since data modeling and records triggers provide arrays of items, design your workflow to handle multiple items appropriately:

<Tabs>
  <Tab title="Option 1: Process all items in a single function">
    ```json theme={"languages":{"custom":["/_languages/kuiper.json","../_languages/kuiper.json"]}}
    {
      "externalId": "process-all-items",
      "type": "function",
      "parameters": {
        "function": {
          "externalId": "batch-processor",
          "data": {
            "items": "${workflow.input.items}"
          }
        }
      }
    }
    ```
  </Tab>

  <Tab title="Option 2: Use dynamic tasks to process items individually">
    ```json theme={"languages":{"custom":["/_languages/kuiper.json","../_languages/kuiper.json"]}}
    {
      "externalId": "generate-individual-tasks",
      "type": "function",
      "parameters": {
        "function": {
          "externalId": "task-generator",
          "data": {
            "items": "${workflow.input.items}"
          }
        }
      }
    },
    {
      "externalId": "process-individual-items",
      "type": "dynamic",
      "parameters": {
        "dynamic": {
          "tasks": "${generate-individual-tasks.output.tasks}"
        }
      },
      "dependsOn": ["generate-individual-tasks"]
    }
    ```
  </Tab>
</Tabs>

## Authenticate a trigger

The trigger requires a [nonce](/cdf/access/concepts/authentication_flows_oidc#on-behalf-of-flow) for authentication. This is a temporary token that's used for authentication when the data workflow execution starts. A nonce can be retrieved from the [Sessions API](/api-reference/concepts/20230101/function-calls) when creating a session.

<Info>
  Make sure that the nonce has the necessary capabilities to create a workflow run and perform the tasks in the workflow.
</Info>

## Trigger run history

You can retrieve a trigger's run history. This gives detailed information about each run, such as when it happened, which data workflow it successfully started, or why the trigger run failed. Use this information to monitor trigger performance and troubleshoot issues with automated workflow executions.

The run history includes:

* **Fire time**: When the trigger has executed.

* **Status**: Whether the trigger successfully started a workflow (`success`) or failed (`failed`).

* **Workflow execution ID**: The ID of the workflow execution that was started (if successful).

* **Reason for failure**: Details about why the trigger failed.