---
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"
prereq: You need a NATS project (the default of `abi new project`) and a pipeline in an in-engine module. Without NATS a declared job is logged and not run.
next:
  - /capabilities/build/pipelines/monitor-runs
  - /capabilities/build/modules/module-jobs
---

# Schedule a pipeline

A pipeline has no schedule of its own. Schedule it with a [module job](/user-guide/build/modules/module-jobs): a method on the module, with triggers, that calls `run`. The engine hosts the job, and JetStream fires its triggers.

## Web UI

Not applicable. The System app's Jobs tab shows each job's schedule and runs it by hand (**Run now**), but no screen sets a schedule: triggers are declared in code. Declare the job with Python, then watch it in [Monitor runs](/user-guide/build/pipelines/monitor-runs).

## CLI

Not applicable. No `abi` command creates or schedules a job. Declare it with Python. To run the pipeline once from a terminal, see [Run a pipeline](/user-guide/build/pipelines/run-a-pipeline?via=cli).

## Python

### Declare the job

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)}
```

An in-engine module may use 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`.

### Pick the trigger

- `Cron("0 0 2 * * *", time_zone="UTC")`: six fields, seconds first, or an alias such as `@daily`.
- `Every("1h")` or `Every(timedelta(hours=1))`: a fixed interval, at least one second.
- `OnEvent(...)`: when an ABI event or a NATS message arrives.

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](/user-guide/build/modules/module-jobs#triggers).

### Restart the engine

The engine hosts the job when it starts. It then shows in the System app's Jobs tab.

## REST API

Not applicable. No route sets a schedule. `POST /api/admin/system/jobs/<module>/<job>/trigger` starts one run of a declared job, and `POST /pipelines/<name>` runs the pipeline once ([Run a pipeline](/user-guide/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. The only job settings are process-wide, under `nats.jobs` (`enabled: false` keeps a process from hosting any job).

## Make it safe to run twice

Delivery is at least once: after a crash or a lost acknowledgement, the handler can run again. A missed cron tick fires once when the broker comes back. Make the pipeline idempotent: write each entity under a stable IRI, so a second run adds no new triples.

:::warning Legacy: Dagster
Without NATS, a declared job is logged and not run. Those projects can schedule a pipeline with a Dagster asset (`abi new orchestration`), a [legacy](/user-guide/build/applications/dagster) extra removed in the next major.
:::