Migrating from plain Python scripts and cron

The most common starting point for adopting a workflow engine is not another workflow engine. It is a folder of Python scripts called from cron, a CI pipeline, or a long-suffering shell script. The honest version of this page is: most of those scripts should stay scripts. A short job that runs at 3am, writes a row to a table, and emails you if it fails is well-served by cron + python script.py + a wrapping shell script that catches non-zero exits. Do not move that to Flux.

The scripts worth moving are the ones that hurt. You can usually identify them with one question: when this script fails halfway through, what do you do? If the answer is “re-run it from the top, hope the first half was idempotent, manually clean up if not” — that is the pain Flux is designed to remove. If the answer is “we have a rerun script, it works, we don’t really think about it” — leave it alone.

This page is the most-trafficked of the migration guides because the pattern is the most common. It is also the one where the honest answer is most often “you don’t need this yet.”

Why move

A few smells that mean you have outgrown the script-and-cron pattern:

• Rerun scripts. Anywhere your team has written a “if the main script crashed, run this one to pick up where it left off” — you have written half of a durable executor by hand.
• Idempotency anxiety. You are not sure whether step 4 ran before the crash. You can find out, but it takes 20 minutes of querying logs. That uncertainty is the cost of not having an event log.
• Cron with locks. Two boxes running the same cron entry have produced a duplicate run twice this year. You added a Postgres advisory lock or a Redis SETNX. The lock is now a thing you maintain.
• Observability holes. “What was the input to last Tuesday’s run?” requires reconstructing it from logs. Sometimes you can; sometimes you can’t.
• Manual retries on flaky APIs. Your scripts have for _ in range(3): try: ... except: time.sleep(backoff); continue. The retry logic is now a thing you read carefully whenever you touch the script.
• Pipelines stapled together with bash. Three scripts piped together, with a bash wrapper that decides which to skip on rerun. The bash wrapper is the workflow engine.

Any one of these is fine. Two or three of them on the same job is the signal. That job is the one to migrate first.

Mental-model translation

Code-pattern translation

A daily batch script.

# Before — daily_report.py invoked from crontab: 0 3 * * * python daily_report.py
import requests, smtplib, logging

def main():
    logging.info("starting")
    rows = fetch_rows()
    logging.info("fetched %d rows", len(rows))
    summary = summarize(rows)
    send_email(summary)
    logging.info("done")

if __name__ == "__main__":
    main()

# After
from flux import cron, ExecutionContext
from flux.task import task
from flux.workflow import workflow

@task
async def fetch_rows() -> list[dict]: ...

@task
async def summarize(rows: list[dict]) -> str: ...

@task
async def send_email(summary: str) -> None: ...

@workflow.with_options(name="daily_report", schedule=cron("0 3 * * *", timezone="UTC"))
async def daily_report(ctx: ExecutionContext[None]):
    rows = await fetch_rows()
    summary = await summarize(rows)
    await send_email(summary)

Three things changed. The body is now an async function. Each step is a @task, so the engine records its input and output. The schedule moved from crontab to the workflow itself — and unlike the “cron with locks” pattern above, firing exactly once is the engine’s job: the scheduler coordinates across server replicas through PostgreSQL advisory locks and persists its run state, so you don’t maintain the lock yourself.

A pipeline of scripts.

# Before — pipeline.sh
python extract.py > /tmp/raw.json
python transform.py < /tmp/raw.json > /tmp/clean.json
python load.py < /tmp/clean.json

# After
from flux import ExecutionContext
from flux.task import task
from flux.tasks import pipeline
from flux.workflow import workflow

@task
async def extract(_: None) -> list[dict]: ...

@task
async def transform(rows: list[dict]) -> list[dict]: ...

@task
async def load(rows: list[dict]) -> int: ...

@workflow
async def etl(ctx: ExecutionContext[None]):
    return await pipeline(extract, transform, load, input=ctx.input)

The intermediate /tmp files disappear. pipeline from flux.tasks threads each task’s output into the next task’s input. If load crashes after transform finished, the next worker resumes at load with the recorded output of transform — no rerun script needed.

Fan-out and aggregate.

# Before — process a list of items
def main():
    items = fetch_items()
    results = []
    for item in items:
        try:
            results.append(process(item))
        except Exception as e:
            logging.warning("item %s failed: %s", item.id, e)
    summary = aggregate(results)
    write(summary)

# After — concurrent, durable, with task-level retries
from flux.tasks import parallel

@task.with_options(retry_max_attempts=3)
async def process(item: dict) -> dict: ...

@task
async def aggregate(results: list[dict]) -> dict: ...

@workflow
async def fan_out(ctx: ExecutionContext[list[dict]]):
    items = ctx.input
    results = await parallel(*(process(item) for item in items))
    return await aggregate(results)

Three wins in one rewrite. The items are processed concurrently instead of sequentially. Each item gets independent retries on flaky API calls. The aggregation step has a recorded input you can inspect later (flux execution show <id> --detailed).

Talking to a flaky API.

# Before — manual retry loop
def call_api(arg):
    for attempt in range(3):
        try:
            return requests.get(url, params={"x": arg}, timeout=10).json()
        except (requests.Timeout, requests.ConnectionError):
            time.sleep(2 ** attempt)
    raise RuntimeError("api failed")

# After
@task.with_options(retry_max_attempts=3, retry_delay=2)
async def call_api(arg: str) -> dict:
    async with httpx.AsyncClient(timeout=10) as client:
        response = await client.get(url, params={"x": arg})
        response.raise_for_status()
        return response.json()

The retry loop becomes a decorator argument. Each retry is recorded as an event, so you can see in the event log how many attempts the call took. If the worker dies between attempts, the next worker picks up the retry from its recorded position.

Producing a large file.

# Before — write to S3, return the URI
def render_report(rows: list[dict]) -> str:
    body = build_huge_report(rows)
    s3.put_object(Bucket="reports", Key=f"{uuid.uuid4()}.pdf", Body=body)
    return uri

# After — let Flux offload the output to object storage
from flux.output_storage import LocalFileStorage  # or a custom OutputStorage subclass for S3

# LocalFileStorage takes no constructor arguments; its base path is
# `<flux.home>/<flux.local_storage_path>`, set in flux.toml.
@task.with_options(output_storage=LocalFileStorage())
async def render_report(rows: list[dict]) -> bytes:
    return build_huge_report(rows)

The task returns the bytes; Flux writes them to the configured output_storage instead of inlining them into the event log. The event log retains a reference, and ctx.output resolves the reference on read. You stop hand-rolling the “write large outputs out-of-band” pattern.

What you give up

The most expensive thing you give up is the simplicity of python script.py. A script needs Python, the script file, and credentials. Flux needs a server process, at least one worker process, and a real database (Postgres in production, SQLite for development). That is more infrastructure to operate, monitor, and back up. If your scripts run on a single VM with no operational team behind them, this matters; if you already run a service plus a database, the delta is small.

You give up the option of running anywhere instantly. A standalone script runs on a developer’s laptop with python script.py. A Flux workflow runs against a Flux server. For testing during development, workflow.run(input) runs the workflow in-process without a server — this is fine for development and for unit tests, but the production execution model assumes a server is present.

You give up the comfort of “I can read this script top-to-bottom.” A Flux workflow is structurally similar — a function with sub-calls — but the determinism contract (time, randomness, and I/O move into tasks) is a real rule and you will trip over it the first time you write a workflow. Plan for a day’s worth of getting comfortable.

Migration order

Pick exactly one job. The one that hurts the most — the multi-step nightly pipeline, the long-running scraper, the report generator that occasionally crashes with no clear restart point. Not the easy one, not the most important one, the painful one. The migration is paying off only if it removes pain, and you cannot measure that on a job that wasn’t hurting.

Wrap the script in a @workflow, even if it has only one step. Break the steps into @tasks. Run it locally first (workflow.run(input)) and confirm the output matches the script’s output for known inputs. Register it against a development Flux server and run it once via the CLI (flux workflow run). Inspect the event log (flux execution show <id> --detailed) and make sure you understand what was recorded.

Then run it in production next to the original cron entry. Pick a different minute so the two runs are distinguishable. Compare outputs daily for one or two weeks. The Flux version is paying off when it survives an event the original script wouldn’t have — a worker restart during the run, an external API timeout, a partial database outage. If you can see in the event log that the Flux version handled it visibly better, you have a real win. Disable the original cron entry. Move on.

If after the first migration the answer is “the Flux version is fine but no better,” stop. The investment isn’t returning. Your scripts and cron were probably good enough.

Honest disqualifier

A single-step script that runs once a day and is fine if it fails — cron-plus-alerting handles it well — does not need durable execution. The runtime cost of adding Flux to that job (one more thing to keep up, one more place a deploy can break) is not zero, and the durability win is not present. The honest engineering answer is to leave it alone.

The same applies to: scripts that are short and idempotent by construction (a few seconds, no external state), one-off data-fixing scripts that run twice in their lifetime, and CI jobs that already have re-run-on-failure built into the CI system.

A workflow engine is a real tool, not a category to migrate to on principle.

Where to read more

• Concepts — vs. plain Python + queues — the comparative analysis behind this guide.
• Get started — your first workflow — the tutorial path.
• Build — Scheduling workflows — the schedule API in detail.
• Build — Errors and retries — the retry option surface.
• Build — Output storage — offloading large outputs.

As of July 2026. Flux 0.56.0.