Routines are workflows that start automatically from dataset, schedule, or completion triggers. Use routines for recurring jobs such as embedding newly appended rows, scheduled evaluation, retraining after a data refresh, or chaining work after another workflow succeeds.

Build a routine by extending MixRoutine and declaring trigger for routine in the run() signature.

Quick Start

from mixtrain import MixRoutine, on_added_rows


class EmbedNewRows(MixRoutine):
    def run(
        self,
        new_rows=on_added_rows("photos", batch_rows=100),
        batch_size: int = 256,
    ):
        # `new_rows` is a read-only Dataset of the rows added/changed since the last run
        for batch in new_rows.batches():
            print(f"Process {batch.num_rows} new rows")

Create it with the routine command:

mixtrain routine create ./embed-new-rows --name embed-new-rows

This will create a new routine, which will trigger when rows are added to or changed in the photos dataset. You can view the routine, run history, and run it manually via the CLI, SDK or the web app.

Trigger Types

Additions or Changes to Dataset Rows

Dataset triggers fire when rows are added to or changed in a dataset. Declare them with on_added_rows(). The trigger parameter receives a read-only Dataset scoped to just the rows added or changed since the previous firing. You can use the dataset to select, filter, to_pandas, or stream with batches().

from mixtrain import MixRoutine, on_added_rows


class IndexDocuments(MixRoutine):
    def run(
        self,
        rows=on_added_rows("documents", batch_rows=500),
        index_name: str = "main",
    ):
        for batch in rows.batches():
            upsert_to_index(index_name, batch)

batch_rows delays firing until at least that many new rows have accumulated. added_fraction can also be used to fire after a fraction of the dataset has been added:

rows=on_added_rows("training-data", added_fraction=0.05)

Deletions from Dataset Rows

It fires when rows are deleted from a dataset (for example, via Dataset.delete(predicate)). Declare with on_deleted_rows() and the trigger parameter receives a read-only Dataset of the removed rows since the previous firing.

from mixtrain import MixRoutine, on_deleted_rows


class ReindexOnDelete(MixRoutine):
    def run(self, removed=on_deleted_rows("documents")):
        drop_from_index(removed.select(["id"]).to_pandas())

Schedules

Schedule triggers run from a cron expression or an interval. Provide exactly one of cron or every.

from mixtrain import MixRoutine, on_schedule


class NightlyEval(MixRoutine):
    def run(self, trigger=on_schedule("0 3 * * *"), model: str = "candidate"):
        run_eval(model)

class HourlyHealthCheck(MixRoutine):
    def run(self, trigger=on_schedule(every="1h")):
        check_outputs()

every accepts one or more integer duration parts with units: s seconds, m minutes, h hours, d days, or w weeks. Examples: 10s, 30m, 1h, 2d, 1w, 1h30m. Timezone is set with tz on on_schedule and defaults to UTC.

Completion

Completion triggers run after another resource reaches a terminal status.

from mixtrain import MixRoutine, on_workflow_success


class EvaluateAfterTrain(MixRoutine):
    def run(self, trigger=on_workflow_success("train-model"), eval_name: str = "regression"):
        print(f"Training run {trigger.run_number} finished")
        start_eval(eval_name, trigger.outputs)

Use on_workflow_failure("name") for failure handling, or on_complete("name", status="any") for the generic form.

Routine Contract

A MixRoutine class follows the MixFlow lifecycle (setup(), run(), cleanup()), with these additional rules:

from mixtrain import MixRoutine, on_schedule


class GoodRoutine(MixRoutine):
    def run(self, trigger=on_schedule(every="1h"), limit: int = 100):
        ...

The platform creates trigger runs with the configured defaults. Manual runs from the app use the same event payload shape and mark the source as manual.

Reading New Dataset Rows

For on_added_rows routines, the trigger parameter is a read-only Dataset of the rows added or changed since the last run — genuinely appended rows, plus rows whose values changed via update() (a change surfaces as its new value).

def run(self, new_rows=on_added_rows("photos")):
    df = new_rows.select(["id", "url"]).to_pandas()   # materialize
    for batch in new_rows.batches():                  # or stream
        process(batch)

    # `new_rows.name`, `.version`, and `.from_version` identify the source
    # dataset and the version range this firing covers:
    full = Dataset(new_rows.name, version=new_rows.version)

Because it is read-only, a routine cannot accidentally mutate the dataset that triggered it.

If a threshold is not met the rows keep accumulating, so each firing sees every row added or changed since the previous successful run.

Entrypoints

If a directory contains multiple routines, pass an entrypoint hint:

mixtrain routine create . --name embed-new-rows --entrypoint routines/embed.py:EmbedNewRows
mixtrain routine create . --name nightly-eval --entrypoint NightlyEval

When a directory contains only MixRoutine classes and no MixFlow classes, Mixtrain can auto-detect the routine entrypoint.

Crash Loop Protection

Completion triggers include loop protection so a chain of routines cannot recurse indefinitely. Repeated dispatch failures can disable a trigger until the routine is updated.

Next Steps

• MixRoutine Reference - Trigger constructors and event fields
• Routine CLI Reference - Create and manage routines
• Workflows - Standard workflow lifecycle
• Datasets - Dataset versions and incremental reads