Orchestrate Excel Reports with Prefect

A report script that runs from cron works until the day it does not, and then nobody knows until someone asks where the Monday report is. The database was restarting at 07:00, the script failed on its first query, cron did not retry, and the log file said so to no one. Prefect is a Python workflow tool that wraps an existing script with very little change: decorate the functions, and each run gets retries, a record of what happened, parameters, alerts and a schedule. Unlike Airflow, a Prefect flow is just Python that can be run and tested like any other function. This page, part of Scheduling Python Excel Scripts with Cron, converts a small weekly report into a flow.

Prerequisites

pip install prefect pandas openpyxl

The examples use Prefect 3. A flow can run as a plain Python call with no server set up — Prefect starts a temporary local API in the background for the run's records — which is how the code below ran. For scheduled runs you either keep a small Prefect server running yourself or use Prefect Cloud. sales_source.csv stands in for the database the real report reads.

Step 1 — Split the script into tasks

Each step that can fail independently becomes a @task; the function that runs them in order becomes the @flow. The code inside the functions is the same as in the original script:

from datetime import date
from pathlib import Path
import pandas as pd
from prefect import flow, task

@task(retries=3, retry_delay_seconds=[1, 2, 4])
def extract(report_date: date) -> pd.DataFrame:
    df = pd.read_csv("sales_source.csv", parse_dates=["Week"])
    return df[df["Week"] == pd.Timestamp(report_date)]

@task
def build(rows: pd.DataFrame, report_date: date) -> Path:
    out = Path("reports") / f"weekly_sales_{report_date:%Y-%m-%d}.xlsx"
    out.parent.mkdir(exist_ok=True)
    rows.sort_values("Sales", ascending=False).to_excel(out, index=False)
    return out

@task(retries=2, retry_delay_seconds=5)
def publish(path: Path) -> Path:
    share = Path("share")
    share.mkdir(exist_ok=True)
    target = share / path.name
    target.write_bytes(path.read_bytes())
    return target

@flow(name="weekly-sales-report")
def weekly_sales_report(report_date: date = date(2026, 9, 28)) -> Path:
    rows = extract(report_date)
    if rows.empty:
        raise ValueError(f"no source rows for {report_date}")
    return publish(build(rows, report_date))

result = weekly_sales_report()
print(result, pd.read_excel(result)["Region"].tolist())

Calling the flow runs it: the tasks execute in order, their results pass from one to the next as ordinary Python values, and Prefect records each run. retry_delay_seconds accepts a list, giving exponential backoff — 1, 2 and then 4 seconds between the three retries of the extract. The empty-result check raises inside the flow, which marks the run as failed; a report built from no rows is a failure, not a success with an empty sheet.

Step 2 — Retries where failures are transient

The value of retries shows when a dependency is briefly unavailable. Simulate a database that refuses the first connection and check what the run did:

attempts = {"n": 0}

@task(retries=3, retry_delay_seconds=0)
def flaky_extract(report_date: date) -> pd.DataFrame:
    attempts["n"] += 1
    if attempts["n"] < 3:
        raise ConnectionError("database is restarting")
    return extract.fn(report_date)

@flow
def report_with_flaky_source(report_date: date = date(2026, 9, 28)) -> int:
    return len(flaky_extract(report_date))

state = report_with_flaky_source(return_state=True)
print(state.type.value, "rows:", state.result(), "attempts:", attempts["n"])

The task failed twice and succeeded on the third attempt, and the flow completed — the Monday report goes out despite the restart. Each failed attempt is logged with its exception, so the run history still shows that the database was unavailable at 07:00, which is useful when the same thing starts happening every week. extract.fn calls the undecorated function inside another task. return_state=True returns the run's final state instead of its result, which is how tests check whether a flow completed, failed or crashed without catching exceptions. Put retries on the steps that talk to other systems — databases, APIs, file shares — and not on steps that fail for reasons a retry cannot fix, such as a missing column.

Step 3 — Parameters make reruns and backfills easy

Because the report date is a parameter, rerunning last week or backfilling a range is a loop over dates, and each run is recorded separately:

from datetime import timedelta

for weeks_back in (2, 1):
    d = date(2026, 9, 28) - timedelta(weeks=weeks_back)
    print(d, "->", weekly_sales_report(report_date=d).name)
print(sorted(p.name for p in Path("share").iterdir()))

A flow whose date comes from date.today() inside the code can only ever produce today's report; a parameter with a default does the same on schedule and anything else on request. Backfill Excel Reports for Past Dates covers backfills in general.

Step 4 — Alert when a run finally fails

Retries handle transient failures; permanent ones need a person. A flow-level on_failure hook runs after the flow has failed for good, with the flow run and its final state:

alerts = []

def notify_failure(flow, flow_run, state):
    alerts.append(f"{flow.name} failed for {flow_run.parameters.get('report_date')}: {state.message}")

@flow(name="weekly-sales-report", on_failure=[notify_failure])
def guarded_report(report_date: date = date(2026, 9, 28)) -> Path:
    return weekly_sales_report.fn(report_date)

state = guarded_report(report_date=date(2026, 10, 5), return_state=True)
print(state.type.value)
print(alerts[0][:110])

There are no source rows for 5 October yet, so the flow fails and the hook records a message naming the flow, the parameter and the error. In a real job the hook posts to Slack or Teams or sends an email — Send an Alert When an Excel Report Job Fails shows those. Prefect's own notification automations can do the same from the server side without code, which is useful when many flows share one alerting policy.

Step 5 — Put it on a schedule

The simplest way to schedule a flow is to serve it: a long-running process that registers the flow with a cron schedule and runs it when due. Run it under systemd, in a container, or anywhere a service can stay up:

if __name__ == "__main__":
    weekly_sales_report.serve(
        name="weekly-sales-monday",
        cron="0 7 * * 1",                # 07:00 every Monday
        parameters={},                   # defaults; the date is computed below
    )

A scheduled run should use the run's date rather than a fixed default; change the parameter's default to None and compute the date inside the flow when it is missing. For fleets of flows, deploy them to a work pool with flow.deploy(...) instead, and let workers — processes, containers or cloud jobs — pick up runs. The UI then shows every run with its parameters, logs, retries and final state, which is the record that cron never had.

The last row is the honest cost: a flow needs somewhere for its runs to be recorded and something to execute them on schedule. For one report that means a small server process; for a team with a dozen reports, that overhead is quickly repaid by having every run, retry and failure in one place.

Common pitfalls

• Retrying everything. Retries on a step that fails because of bad data only delay the alert; put them on steps that call other systems.
• Dates computed at import time. A default like date.today() in a signature is evaluated once when the module loads; compute inside the flow.
• Returning huge objects between tasks. Results are kept in memory and may be persisted; pass paths for large DataFrames.
• No alert on final failure. Retries hide transient errors; a hook or automation must surface the permanent ones.
• Testing only through the scheduler. Call the flow directly with return_state=True in tests.

Conclusion

Decorate the report's steps as tasks and the orchestration as a flow, add retries with backoff to the steps that touch databases, APIs and shares, and pass the report date as a parameter so reruns and backfills are ordinary calls. Attach an on_failure hook so a permanent failure reaches a person, test the flow by calling it with return_state=True, and schedule it by serving it with a cron expression or deploying it to a work pool.

Frequently asked questions

How do I schedule a Python Excel report with Prefect? Wrap the script in a @flow, then call flow.serve(name=..., cron="0 7 * * 1") in a long-running process, or deploy it to a work pool with flow.deploy. Prefect then runs it on the schedule and records each run.

How do I add retries to a step in Prefect? Pass retries and retry_delay_seconds to @task, for example @task(retries=3, retry_delay_seconds=1, 2, 4) for exponential backoff. The task is rerun automatically when it raises.

Do I need a Prefect server to run a flow? No, calling a flow function runs it immediately, with Prefect starting a temporary local API for the run records. Scheduled runs need a running Prefect server or Prefect Cloud plus a process that serves or executes the flows.

How is Prefect different from Airflow for reports? Prefect flows are ordinary Python functions that can be called and tested directly, with parameters passed at run time. Airflow is organised around scheduled DAG files and suits large platforms with many interdependent pipelines.

Related

• Up one level: Scheduling Python Excel Scripts with Cron — every way to run reports on a schedule.
• Orchestrate Excel Reports with Apache Airflow — the heavier alternative.
• Choose a Scheduler for Python Excel Reports — when Prefect is the right tool.
• Retry a Failed Excel Report Job in Python — retries without an orchestrator.