---
question: How do I run a pipeline on a cron?
kind: task
via_web: na
via_cli: na
via_python: available
via_api: na
via_config: na
ref_python: "@job / DagsterOrchestration"
prereq: You need a pipeline in a module. On ABI 3.x in a NATS project (the default of `abi new project`), a module job runs it. On ABI 2.x, and on 3.x without NATS, a Dagster orchestration runs it.
next:
  - /v2/capabilities/build/pipelines/monitor-runs
  - /v2/capabilities/build/workflows/schedule-a-workflow
  - /v2/capabilities/build/modules/module-jobs
---

> **You are reading the v2.x docs (engine modules). New projects use v3.** [Read this page in v3](/capabilities/build/pipelines/schedule-a-pipeline) · [Migrate from 2.x to 3.0](/capabilities/upgrade/migrate-to-3)

# Schedule a pipeline

A pipeline has no schedule of its own. Something on its module calls `run` on a schedule, and what that is depends on the ABI line:

| Line | What calls `run` | Fired by |
|---|---|---|
| ABI 3.x, NATS project | A [module job](/v2/capabilities/build/modules/module-jobs) (`@job`) or a `@workflow` with triggers, on the module's `ABIModule` | JetStream |
| ABI 2.x, and 3.x without NATS | A Dagster asset in a `DagsterOrchestration` under the module's `orchestrations/` folder | Dagster |

On ABI 3.x without NATS, a declared `@job` is logged and not run. ABI 2.x has no `@job` at all.

## Web UI

Not applicable. No screen sets a schedule: triggers are declared in code. On ABI 3.x in a NATS project, the System app's Jobs tab shows each job's schedule and runs it by hand (**Run now**): see [Monitor runs](/v2/capabilities/build/pipelines/monitor-runs). Dagster has its own UI. Declare the schedule with Python.

## CLI

Not applicable. No `abi` command schedules a job. `abi new orchestration` scaffolds an empty Dagster orchestration (see the Python steps). To run the pipeline once from a terminal, see [Run a pipeline](/v2/capabilities/build/pipelines/run-a-pipeline?via=cli).

## Python

### Declare a module job

ABI 3.x, in a NATS project. On the `ABIModule` of the module that owns the pipeline:

```python title="src/my_module/__init__.py"
from datetime import timedelta

from naas_abi_core.module.Module import BaseModule
from naas_abi_core.module.jobs import Cron, JobContext, job


class ABIModule(BaseModule):
    @job(triggers=(Cron("0 0 2 * * *", time_zone="UTC"),), timeout=timedelta(hours=1))
    def nightly_ingest(self, ctx: JobContext) -> dict:
        from my_module.pipelines.ItemsPipeline import (
            ItemsPipeline,
            ItemsPipelineConfiguration,
            ItemsPipelineParameters,
        )

        graph = ItemsPipeline(ItemsPipelineConfiguration()).run(ItemsPipelineParameters())
        ctx.log(f"inserted {len(graph)} triples")
        return {"triples": len(graph)}
```

The handler may be a plain function, which runs in a worker thread. Return something JSON-serializable (a count, not the graph): it is stored on the run record. The graph is already stored by `insert` inside `run`.

### Or declare a `@workflow`

ABI 3.x, in a NATS project. A `@workflow` with triggers adds an input to each tick, checked against its input model, and stores the result of each step:

```python title="src/my_module/__init__.py"
from naas_abi_core.module.jobs import Cron
from naas_abi_core.module.workflows import SyncFlowContext, workflow


class ABIModule(BaseModule):
    @workflow(triggers=(Cron("0 0 2 * * *", input={"filter": "open"}),))
    def nightly_ingest(self, ctx: SyncFlowContext, inp: dict) -> dict:
        from my_module.pipelines.ItemsPipeline import (
            ItemsPipeline,
            ItemsPipelineConfiguration,
            ItemsPipelineParameters,
        )

        pipeline = ItemsPipeline(ItemsPipelineConfiguration())
        triples = ctx.step("ingest", lambda: len(pipeline.run(ItemsPipelineParameters(**inp))))
        return {"triples": triples}
```

### Pick the trigger

For a module job or a `@workflow`:

- `Cron("0 0 2 * * *", time_zone="UTC")`: six fields, seconds first, or an alias such as `@daily`. The aliases and time zones are on [Schedule a job](/v2/capabilities/build/orchestrations/schedule-a-job?via=python).
- `Every("1h")` or `Every(timedelta(hours=1))`: a fixed interval, at least one second.
- `OnEvent(...)`: when an ABI event or a NATS message arrives. See [React to an event](/v2/capabilities/build/orchestrations/react-to-an-event?via=python).

A job takes up to 16 triggers, plus `timeout`, `max_attempts` (default 1) and `max_concurrency` (default 1, so a run never overlaps the next). The table is on [Module jobs](/v2/capabilities/build/modules/module-jobs#triggers). `pipeline.<slug>` must not also be the name of a `@job` or a `@workflow` of the module. More on triggers and startup errors: [Schedule a workflow](/v2/capabilities/build/workflows/schedule-a-workflow?via=python).

### Or write a Dagster orchestration

ABI 2.x, and 3.x without NATS. Scaffold the class in the module's `orchestrations/` folder:

```bash
abi new orchestration Items orchestrations
```

Then wrap the pipeline in an asset and schedule it:

```python title="orchestrations/ItemsOrchestration.py"
import dagster as dg
from naas_abi_core.orchestrations.DagsterOrchestration import DagsterOrchestration


@dg.asset(name="items", group_name="my_module")
def items(context: dg.AssetExecutionContext) -> None:
    from my_module.pipelines.ItemsPipeline import (
        ItemsPipeline,
        ItemsPipelineConfiguration,
        ItemsPipelineParameters,
    )

    graph = ItemsPipeline(ItemsPipelineConfiguration()).run(ItemsPipelineParameters())
    context.log.info(f"inserted {len(graph)} triples")


nightly_items = dg.ScheduleDefinition(
    job=dg.define_asset_job("nightly_items", selection=["items"]),
    cron_schedule="0 2 * * *",
)


class ItemsOrchestration(DagsterOrchestration):
    @classmethod
    def New(cls) -> "ItemsOrchestration":
        return cls(definitions=dg.Definitions(assets=[items], schedules=[nightly_items]))
```

A Dagster cron has five fields, minutes first. On ABI 3.x, Dagster is a legacy extra you install yourself: see [Dagster (legacy)](/v2/capabilities/build/applications/dagster).

### Restart the module

On ABI 3.x in a NATS project, the engine hosts the schedule when it starts, and the job shows in the System app's Jobs tab. With Dagster, restart Dagster (in a project without NATS, `abi dev up` starts it), and the schedule shows in its UI.

## REST API

Not applicable. No route sets a schedule. `POST /pipelines/<name>` runs the pipeline once on ABI 2.x and 3.x, and, on ABI 3.x in a NATS project, `POST /api/workflows/<module>/pipeline.<slug>/runs` starts one recorded run ([Run a pipeline](/v2/capabilities/build/pipelines/run-a-pipeline?via=api)). An outside scheduler can call either.

## Config

Not applicable. `config.yaml` has no per-job key: triggers live in code. On ABI 3.x in a NATS project, the only job settings are process-wide, under `nats.jobs` (`enabled: false` keeps a process from hosting any job), and `nats.workflows.bridge` (publishing the engine's pipelines as `pipeline.<slug>`).

## Make it safe to run twice

On NATS, delivery is at least once: after a crash or a lost acknowledgement, the handler can run again, and a missed cron tick fires once when the broker comes back. A Dagster run can be re-executed too. Make the pipeline idempotent: write each entity under a stable IRI, so a second run adds no new triples.