Build a Resilient Celery Pipeline

Build a Celery and Redis pipeline that runs, retries, and summarizes reports.

Introduction

30 Second Summary

Slow reports have a habit of freezing the same screen you need for your next task. A single temporary outage can also leave several reports half-finished.

In this project, you will build a local report pipeline that moves slow work into the background with Celery and Redis. You will watch parallel reports recover from a temporary failure through an automatic retry.

What You'll Build

You will submit a report from one terminal to receive a task ID immediately while a background worker creates the finished CSV file in another.

By the end of this project, you'll have:

  • A background submission flow that gives you a task ID immediately while report work continues.
  • A status command that shows PENDING, RETRY, FAILURE, or SUCCESS before printing the finished result.
  • A parallel three-report batch with completed orders, inventory, and customer CSV files after automatic recovery from a temporary outage.
  • Secret Mission: Add a chord that writes one JSON batch summary only after every report finishes.

Are there any prerequisites?

This project assumes you are comfortable reading Python. You need a Mac with Homebrew, Visual Studio Code, and Redis already installed.

Before We Start

This checkpoint commits you to a local report pipeline powered by Celery with Redis. Slow report generation leaves the foreground CLI so each command can return immediately.

Prepare the Local Celery Environment

Your report CLI will hand slow work to Celery. Redis will carry each queued job to a worker.

Environment problems can look like task bugs. This step isolates Celery 5.6.3 inside .venv. You will also prove the existing Redis service is reachable before writing any worker code.

In this step, get ready to:
  • Create the project directory with an isolated Python environment.
  • Install the pinned Celery bundle with Redis support.
  • Confirm the local services respond through their command-line checks.
Create the project environment

A virtual environment keeps this project's packages separate from your system Python. It also ties the project to a specific Python interpreter.

  • Press Cmd+Space to open Spotlight.
  • Type Terminal into Spotlight.
  • Press Enter to open macOS Terminal.
  • Set up the celery-report-pipeline directory on your Desktop by running these commands:
cd ~/Desktop
mkdir celery-report-pipeline
cd celery-report-pipeline

What do these commands do?

  • The first command points Terminal at your Desktop.
  • The second command creates the celery-report-pipeline directory.
  • The final command makes that directory your current location.

Your Terminal prompt should now show that you are inside celery-report-pipeline.

  • Build the active .venv environment by running these commands:
python3 -m venv .venv
source .venv/bin/activate

How does the virtual environment work?

The first command creates an isolated Python environment inside .venv. The second command updates the current Terminal session to use it.

You should see (.venv) near the start of your Terminal prompt. That prefix confirms the environment is active.

Celery 5.6.3 requires Python 3.9 or newer. The version check tells you whether this environment meets that requirement.

  • Check the active environment's Python version by running:
python --version

What does this check prove?

The command reports the Python interpreter selected by the active virtual environment. A version of 3.9 or newer can run Celery 5.6.3.

✔️ I see version 3.9 or higher

Your Python environment meets Celery's requirement. That removes the first source of setup problems.

ⓧ I see an older version

The current .venv uses an older Python interpreter. Recreate it after Homebrew installs the current Python formula.

  • Install the current Homebrew Python by running:
brew install python@3.14

What does this install?

Homebrew installs Python 3.14 on your Mac. The existing .venv still points to the older interpreter until you recreate it.

  • Use Finder to delete .venv from the celery-report-pipeline directory.
  • Recreate the active environment by running these commands in the same Terminal:
python3 -m venv .venv
source .venv/bin/activate

Why recreate the environment?

A virtual environment keeps the interpreter chosen when it is created. Rebuilding .venv connects the project to the newer Python installation.

  • Confirm the replacement environment meets the requirement by running:
python --version

What should the version confirm?

The active environment should now report Python 3.9 or newer. You can continue once that requirement is satisfied.

ⓧ Command not found

The system cannot find the Python command needed by this project. Homebrew can install the current Python formula before you rebuild .venv.

  • Install the current Homebrew Python by running:
brew install python@3.14

What does Homebrew add?

Homebrew adds Python 3.14 to your Mac. This version meets Celery's minimum requirement.

  • Use Finder to delete any incomplete .venv directory from celery-report-pipeline.
  • Create the active environment by running these commands in the project Terminal:
python3 -m venv .venv
source .venv/bin/activate

What happens now?

Python creates a fresh environment inside .venv. Activation makes its interpreter available as python in this Terminal.

  • Check the newly activated interpreter by running:
python --version

What should the check show?

The command should report Python 3.9 or newer. The (.venv) prefix should remain visible in your prompt.

Having trouble with the environment?

  • Confirm Terminal is still inside the celery-report-pipeline directory before creating .venv.
  • Repeat the activation command if the (.venv) prefix is missing from your prompt.

Still stuck? Help me troubleshoot my Python virtual environment.

Install Celery with Redis support

The Redis bundle gives Celery the support needed to use Redis as a message broker or result backend. Pinning the package to 5.6.3 keeps every later command aligned with this project.

  • Install the pinned Celery Redis bundle into .venv by running:
python -m pip install "celery[redis]==5.6.3"

What does this command install?

  • The python -m pip portion uses the package installer from your active virtual environment.
  • The celery[redis] bundle adds the packages Celery needs for Redis communication.
  • The ==5.6.3 pin installs the exact Celery release used throughout this project.

The installation output should finish without an error. Your active environment now contains the task queue tooling needed for the report pipeline.

Seeing an installation error?

  • Check that (.venv) appears in the Terminal prompt before repeating the installation.
  • Confirm the active Python version is 3.9 or newer.
  • Check your internet connection if the package download cannot complete.

Need another pair of eyes? Help me fix the Celery installation.

Verify Redis and Celery

The worker will depend on the existing Redis service at localhost:6379. A health check separates Redis availability from any Celery code you write later.

Before you run the check, what do you expect Redis to return when it is reachable?

  • Test the local Redis service by running:
redis-cli ping

What does this command check?

The Redis command-line client sends a health check to the local service on port 6379. Its reply shows whether Redis is accepting connections.

✔️ I see PONG

Good work. PONG confirms that the existing Redis service is running at localhost:6379.

ⓧ Redis is not responding

The existing Redis installation is currently stopped. Homebrew can start the service without reinstalling it.

  • Start the existing Redis service by running:
brew services start redis

What does this command change?

Homebrew starts Redis now. It also registers the service to restart when you log in.

  • Repeat the Redis health check by running:
redis-cli ping

What should Redis return now?

You should see PONG in Terminal. That reply confirms Redis is ready for Celery.

The final check confirms that the activated environment can find Celery. Before you run it, what kind of information do you expect a command help screen to contain?

  • Display Celery's command help from the active environment by running:
celery --help

What does this final check prove?

The help screen confirms the celery command is installed inside the active environment. It should list the command's available options plus subcommands.

You should see Celery's command help in Terminal. Together with PONG, this proves the task tooling can reach its local broker.

✔️ Awesome, I've got everything!

Your project environment is ready. Keep this Terminal open with (.venv) visible in the prompt.

ⓧ I'd like to double check the full code

There are no application files to compare yet. Double-check the setup state before moving on.

  • Confirm Terminal is inside the celery-report-pipeline directory.
  • Confirm the prompt includes (.venv).
  • Confirm the active Python version is 3.9 or newer.
  • Confirm celery[redis]==5.6.3 is installed in .venv.
  • Confirm the Redis health check prints PONG.
  • Confirm Celery displays its command help.

That is the environment locked in. Next, you will build the synchronous report generator that makes the foreground delay visible.

Build the Blocking Report Generator

Your Celery environment is ready. Redis remains reachable from the project.

A working report gives you a concrete baseline before background processing enters the picture. This version includes a deliberate three-second pause so you can observe how slow report work affects the foreground process.

In this step, get ready to:
  • Define in-memory data for three reports.
  • Build a writer that produces a CSV file after a delay.
  • Run a synchronous CLI to expose the foreground pause.
Define the report data

Each report begins as a small collection of metric names plus values. The writer turns those rows into a CSV file that another person or system can open.

  • Press Cmd+Space on macOS or Windows+S on Windows to open system search.
  • Type Visual Studio Code into the search field.
  • Press Enter to start Visual Studio Code.
  • Use the folder-opening control to load the existing celery-report-pipeline directory.
  • Create reports.py inside celery-report-pipeline from the editor sidebar.
  • Populate reports.py with the imports plus the three in-memory datasets by copying this code:
import os
import time

REPORT_DATA = {
    "orders": [
        ("orders_created", 42),
        ("orders_shipped", 35),
    ],
    "inventory": [
        ("items_in_stock", 128),
        ("items_low_stock", 7),
    ],
    "customers": [
        ("active_customers", 64),
        ("new_customers", 9),
    ],
}

AVAILABLE_REPORTS = tuple(REPORT_DATA)

What does this code do?

  • The os import provides the directory helper used when the output folder is created.
  • The time import provides the deliberate delay used to represent slow report work.
  • REPORT_DATA maps each report name to two metric rows.
  • AVAILABLE_REPORTS derives the accepted report names from the dictionary keys.
  • Save reports.py.
  • Confirm the editor sidebar shows reports.py inside celery-report-pipeline.

Seeing a syntax warning?

  • Check that every opening square bracket in REPORT_DATA has a matching closing bracket.
  • Check that each metric row contains a quoted metric plus its number inside parentheses.
  • Check that AVAILABLE_REPORTS sits below the closing brace for REPORT_DATA.

Still stuck? Help me compare the report data structure with the expected Python syntax.

The data is ready for a reusable writer. write_report() validates the requested name before creating the corresponding file.

  • Add the CSV writer below AVAILABLE_REPORTS in reports.py by copying this code:
def write_report(report_name: str) -> str:
    if report_name not in REPORT_DATA:
        raise ValueError(f"Unknown report: {report_name}")

    time.sleep(3)
    os.makedirs("output", exist_ok=True)
    output_path = f"output/{report_name}.csv"

    with open(output_path, "w", encoding="utf-8") as report_file:
        report_file.write("metric,value\n")
        for metric, value in REPORT_DATA[report_name]:
            report_file.write(f"{metric},{value}\n")

    return output_path

How does the writer work?

  • The validation rejects names that do not exist in REPORT_DATA.
  • time.sleep(3) represents three seconds of slow report processing.
  • os.makedirs() ensures the output directory exists before the file is written.
  • The file-writing loop adds one CSV row for each metric tuple.
  • The return value gives the caller the generated file path.
  • Save reports.py.
  • Confirm write_report() appears below AVAILABLE_REPORTS without syntax diagnostics in the editor.

Writer showing an error?

  • Check that the body of write_report() uses consistent four-space indentation.
  • Check that the newline characters inside the file-writing strings remain \n.
  • Check that return output_path lines up with the validation plus file-writing statements inside the function.

Need another pair of eyes? Help me find the syntax or indentation problem in write_report().

✔️ Awesome, I've got everything!

Good work. reports.py now contains the complete datasets plus the CSV writer.

ⓧ I'd like to double check the full code

Compare your saved reports.py with this complete version.

import os
import time

REPORT_DATA = {
    "orders": [
        ("orders_created", 42),
        ("orders_shipped", 35),
    ],
    "inventory": [
        ("items_in_stock", 128),
        ("items_low_stock", 7),
    ],
    "customers": [
        ("active_customers", 64),
        ("new_customers", 9),
    ],
}

AVAILABLE_REPORTS = tuple(REPORT_DATA)


def write_report(report_name: str) -> str:
    if report_name not in REPORT_DATA:
        raise ValueError(f"Unknown report: {report_name}")

    time.sleep(3)
    os.makedirs("output", exist_ok=True)
    output_path = f"output/{report_name}.csv"

    with open(output_path, "w", encoding="utf-8") as report_file:
        report_file.write("metric,value\n")
        for metric, value in REPORT_DATA[report_name]:
            report_file.write(f"{metric},{value}\n")

    return output_path

How to use this reference

Compare the imports plus data definitions first. Finish by checking the spacing inside write_report().

Build the synchronous CLI

The report writer needs a small command-line interface that accepts a report name. The first version supports only the sync command so its foreground behavior stays easy to observe.

  • Create cli.py inside celery-report-pipeline from the editor sidebar.
  • Add the imports plus synchronous command helpers to cli.py by copying this code:
import sys

from reports import AVAILABLE_REPORTS, write_report

USAGE = """Usage:
  python cli.py sync <report>

Available reports: orders inventory customers
"""


def validate_report(report_name: str) -> str:
    if report_name not in AVAILABLE_REPORTS:
        raise SystemExit(f"Unknown report: {report_name}\n\n{USAGE}")
    return report_name


def run_sync(report_name: str) -> None:
    output_path = write_report(report_name)
    print(f"Created {output_path}")

What does this code do?

  • sys.argv provides the command-line arguments that the routing logic reads.
  • USAGE records the supported command shape plus the accepted report names.
  • validate_report() stops the program when a report name is unknown.
  • run_sync() calls the report writer in the foreground before printing the returned path.
  • Save cli.py.
  • Confirm the editor sidebar lists cli.py beside reports.py.

Imports not resolving?

  • Check that cli.py sits in the same celery-report-pipeline directory as reports.py.
  • Check that the import uses reports without the .py extension.
  • Check that AVAILABLE_REPORTS plus write_report match their spellings in reports.py.

Still seeing an import warning? Help me diagnose why cli.py cannot import from reports.py.

The helpers now need routing logic that checks the command plus its arguments. main() connects the terminal input to run_sync().

  • Complete the CLI by adding this routing logic below run_sync() in cli.py.
def main() -> None:
    if len(sys.argv) < 2:
        raise SystemExit(USAGE)

    command = sys.argv[1]
    arguments = sys.argv[2:]

    if command == "sync" and len(arguments) == 1:
        run_sync(validate_report(arguments[0]))
    else:
        raise SystemExit(USAGE)


if __name__ == "__main__":
    main()

How does command routing work?

  • main() requires at least one command after the script name.
  • command holds the requested operation.
  • arguments holds everything supplied after the operation.
  • The sync branch accepts exactly one report name before running the validated request.
  • The final guard starts main() when Python runs cli.py directly.
  • Save cli.py.
  • Confirm main() appears below run_sync() without syntax diagnostics in the editor.

Routing logic showing an error?

  • Check that the if plus else branches inside main() use consistent indentation.
  • Check that the final main() call sits inside the __main__ guard.
  • Check that arguments[0] uses square brackets around the zero.

Need help checking the route? Help me find the syntax or indentation issue in cli.py.

✔️ Awesome, I've got everything!

Great. cli.py now accepts one synchronous report command.

ⓧ I'd like to double check the full code

Compare your saved cli.py with this complete synchronous version.

import sys

from reports import AVAILABLE_REPORTS, write_report

USAGE = """Usage:
  python cli.py sync <report>

Available reports: orders inventory customers
"""


def validate_report(report_name: str) -> str:
    if report_name not in AVAILABLE_REPORTS:
        raise SystemExit(f"Unknown report: {report_name}\n\n{USAGE}")
    return report_name


def run_sync(report_name: str) -> None:
    output_path = write_report(report_name)
    print(f"Created {output_path}")


def main() -> None:
    if len(sys.argv) < 2:
        raise SystemExit(USAGE)

    command = sys.argv[1]
    arguments = sys.argv[2:]

    if command == "sync" and len(arguments) == 1:
        run_sync(validate_report(arguments[0]))
    else:
        raise SystemExit(USAGE)


if __name__ == "__main__":
    main()

How to use this reference

Compare the imports plus helper functions first. Finish by checking the indentation inside main().

Run and inspect the blocking report

The synchronous path keeps report generation inside the same process that accepts your command. Running it now reveals how the deliberate delay affects that process.

Before you run the command, when do you think the terminal prompt becomes available again?

  • Switch back to the macOS Terminal from earlier.
  • Generate the orders report by running this command:
python cli.py sync orders

What did the baseline reveal?

Your terminal pauses for about three seconds. It then prints Created output/orders.csv.

During the pause, the foreground process cannot accept another command. That delay is the intended limitation of this synchronous baseline.

That is the hard part done: your baseline produces a real report while exposing the delay you will improve.

Report command not working?

  • If Python cannot import reports or cli, confirm the Terminal is still inside celery-report-pipeline.
  • If the CLI reports an unknown report, confirm the final argument is exactly orders.
  • If the command returns immediately without a path, confirm run_sync() calls write_report() before printing.

Still blocked by an error? Help me debug why the synchronous orders report does not create its CSV file.

  • Switch back to Visual Studio Code.
  • Expand the output directory in the editor sidebar.
  • Select orders.csv to inspect the generated report.

You should see metric,value on the first line. The next two lines are orders_created,42 plus orders_shipped,35.

The file confirms that report generation works. The three-second wait confirms that slow work currently holds the CLI in the foreground.

Your synchronous baseline now writes a valid orders report with a visible delay. Next, you will move that slow work to a Celery worker so the CLI can return immediately.

Move Report Work to a Celery Worker

Your synchronous CSV report proved that the generator works. Its three-second delay also kept the CLI occupied until the file was ready.

Celery moves report generation into a background worker. The caller receives a task ID immediately.

Redis serves as the message broker. It also serves as the result backend.

In this step, get ready to:
  • Create a Celery app with a background report task.
  • Extend the CLI for task submission plus status inspection.
  • Run a worker to prove that report generation leaves the foreground process.
Create the Celery app and task

The Celery app tells producers where to send tasks. It also tells status checks where to retrieve task states and return values.

  • Create a new tasks.py file inside the celery-report-pipeline folder using the Visual Studio Code Explorer sidebar.
  • Define the app plus its first background task by copying this code into tasks.py:
from celery import Celery

from reports import write_report

app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/0",
)


@app.task
def generate_report(report_name: str) -> str:
    return write_report(report_name)

What does this code do?

  • The app object configures Redis database 0 as the broker.
  • The same Redis database stores task states plus return values.
  • The @app.task decorator registers generate_report as work a Celery worker can execute.
  • The task passes its report name to the existing write_report() function.
  • Save tasks.py.

You should now see tasks.py beside reports.py in the Explorer sidebar.

Seeing an import warning?

Confirm that tasks.py sits in the same celery-report-pipeline folder as reports.py.

If the Celery import remains unresolved, confirm that Visual Studio Code is using the Python interpreter from .venv.

Still stuck? Help me check the imports and Celery app in tasks.py.

✔️ Awesome, I've got everything!

Your tasks.py file now defines the Celery app plus generate_report.

ⓧ I'd like to double check the full code

from celery import Celery

from reports import write_report

app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/0",
)


@app.task
def generate_report(report_name: str) -> str:
    return write_report(report_name)
Extend the CLI

The CLI needs one command to submit work. It needs another command to retrieve the state later.

  • Add the task imports below the existing report imports in cli.py.
  • Replace the existing USAGE string with the version shown below.
import sys

from reports import AVAILABLE_REPORTS, write_report
from tasks import app, generate_report

USAGE = """Usage:
  python cli.py sync <report>
  python cli.py submit <report>
  python cli.py status <task-id>

Available reports: orders inventory customers
"""

What changed in the CLI?

  • The task imports give the CLI access to the configured app plus the generate_report task.
  • The updated usage text introduces the submit command.
  • It also introduces the status command.
  • Save cli.py.
  • Confirm that the new imports preserve the synchronous command by running:
python cli.py sync orders

What does this check prove?

This command imports the new Celery configuration before following the original synchronous path. A successful report proves that the imports load without breaking existing behavior.

This regression check still pauses for about three seconds. You should then see Created output/orders.csv.

Does the synchronous command fail now?

Check that the task import reads from tasks import app, generate_report.

Confirm that the original import from reports.py is still present.

Need another pair of eyes? Help me debug the new imports in cli.py.

  • Add submit_report() below run_sync() in cli.py by copying this function:
def submit_report(report_name: str) -> None:
    result = generate_report.delay(report_name)
    print(f"Task ID: {result.id}")

How does submission work?

  • The .delay() call sends the report name to the broker.
  • The returned result object contains the task ID before the report finishes.
  • Printing that ID gives the caller a handle for future status checks.
  • Save cli.py.
  • Verify that the new function parses without affecting the existing path by running:
python cli.py sync orders

Why repeat the synchronous check?

Python reads every function definition when the script starts. This run confirms that submit_report() has valid syntax.

The three-second pause remains part of the synchronous baseline. You should see Created output/orders.csv after it finishes.

  • Add show_status() below submit_report() in cli.py by copying this function:
def show_status(task_id: str) -> None:
    result = app.AsyncResult(task_id)
    print(f"State: {result.state}")
    if result.ready():
        value = result.get(propagate=False)
        print(f"Result: {value!r}")

How does status inspection work?

  • The app.AsyncResult(task_id) call creates a result object for the supplied task ID.
  • The .state property reports the current task state.
  • The .ready() check prevents the CLI from requesting an unfinished result.
  • The .get(propagate=False) call retrieves the finished value without re-raising a failed task exception.
  • Save cli.py.
  • Verify that the status function parses by running the synchronous command again:
python cli.py sync orders

What does this run confirm?

This run confirms that Python can load the completed status function. You should see Created output/orders.csv after the familiar three-second pause.

Seeing a syntax error?

Check that the if result.ready(): line ends with a colon.

Confirm that both lines inside the if block use the same indentation.

Still seeing the error? Help me compare my show_status function with the expected code.

  • Replace the command-selection block inside main() with the completed routing logic below:
    if command == "sync" and len(arguments) == 1:
        run_sync(validate_report(arguments[0]))
    elif command == "submit" and len(arguments) == 1:
        submit_report(validate_report(arguments[0]))
    elif command == "status" and len(arguments) == 1:
        show_status(arguments[0])
    else:
        raise SystemExit(USAGE)

How does the routing work?

  • The sync branch keeps the blocking command available for comparison.
  • The submit branch validates the report name before queuing it.
  • The status branch passes the supplied task ID to show_status().
  • Save cli.py.

✔️ Awesome, I've got everything!

Your CLI now supports synchronous generation, background submission, and later status inspection.

ⓧ I'd like to double check the full code

import sys

from reports import AVAILABLE_REPORTS, write_report
from tasks import app, generate_report

USAGE = """Usage:
  python cli.py sync <report>
  python cli.py submit <report>
  python cli.py status <task-id>

Available reports: orders inventory customers
"""


def validate_report(report_name: str) -> str:
    if report_name not in AVAILABLE_REPORTS:
        raise SystemExit(f"Unknown report: {report_name}\n\n{USAGE}")
    return report_name


def run_sync(report_name: str) -> None:
    output_path = write_report(report_name)
    print(f"Created {output_path}")


def submit_report(report_name: str) -> None:
    result = generate_report.delay(report_name)
    print(f"Task ID: {result.id}")


def show_status(task_id: str) -> None:
    result = app.AsyncResult(task_id)
    print(f"State: {result.state}")
    if result.ready():
        value = result.get(propagate=False)
        print(f"Result: {value!r}")


def main() -> None:
    if len(sys.argv) < 2:
        raise SystemExit(USAGE)

    command = sys.argv[1]
    arguments = sys.argv[2:]

    if command == "sync" and len(arguments) == 1:
        run_sync(validate_report(arguments[0]))
    elif command == "submit" and len(arguments) == 1:
        submit_report(validate_report(arguments[0]))
    elif command == "status" and len(arguments) == 1:
        show_status(arguments[0])
    else:
        raise SystemExit(USAGE)


if __name__ == "__main__":
    main()
Run and inspect a background task

A worker is a separate process that waits for queued tasks. Keeping it in another Terminal window leaves your original Terminal free for submissions and status checks.

  • Create a second Terminal window from the Terminal session you already have by pressing Cmd+N.
  • Move into celery-report-pipeline and activate its existing virtual environment by running:
cd ~/Desktop/celery-report-pipeline
source .venv/bin/activate

Why activate the second Terminal?

Each Terminal window has its own shell session. Activating .venv gives the worker access to the installed Celery package.

Before you start the worker, which task name do you expect it to register?

  • Start the worker from the second activated Terminal by running:
celery -A tasks worker --loglevel=INFO

What does this command do?

  • The -A tasks option loads the Celery app from tasks.py.
  • The worker subcommand starts the process that consumes queued tasks.
  • The --loglevel=INFO option displays task activity in the Terminal.

You should see tasks.generate_report in the registered task list. Leave this worker running.

Worker does not start?

Confirm that the second Terminal prompt shows (.venv) before the worker command.

Check that tasks.py is inside the current celery-report-pipeline folder.

Need help with the worker? Help me diagnose why Celery cannot load tasks.py or connect to Redis.

Before you submit the report, do you think the CLI waits three seconds or returns a task ID immediately?

  • Switch back to the original activated Terminal.
  • Submit an orders report by running:
python cli.py submit orders

What happens during submission?

The CLI sends the report name to Redis through generate_report.delay(). It prints the task ID without waiting for write_report() to finish.

You should immediately see a line beginning with Task ID:.

  • Record the printed task ID here: your task ID.
  • Inspect the task immediately by running:
python cli.py status [[TASK_ID="your task ID"]]

What does the first status check show?

The CLI asks the Redis result backend for the current state. You should see PENDING if the report is still underway.

A fast worker may finish before this check. In that case, you see SUCCESS plus the result.

  • Wait until the worker Terminal logs that tasks.generate_report has completed.

Before you check again, which final state do you expect to see now?

  • Inspect the completed task by running the status command again:
python cli.py status [[TASK_ID="your task ID"]]

What proves the task completed?

The result backend now reports State: SUCCESS. The next line shows Result: 'output/orders.csv'.

That is the split working: the CLI submitted the task immediately while the worker handled the slow report generation.

Status stays on PENDING?

Check the second Terminal to confirm the worker remains running. Look for activity from tasks.generate_report.

Confirm that both broker and backend use redis://localhost:6379/0 in tasks.py.

Still waiting? Help me trace why my Celery task remains pending.

Your caller now returns immediately while the worker creates reports in the background. Next, you'll fan several reports out in parallel and expose what happens when one task fails.

Fan Out a Batch and Expose a Failure

Your CLI now sends one report to a Celery worker. The next goal is to submit several reports with one command.

A Canvas group dispatches independent report tasks in parallel through Redis. This can increase throughput.

This step tests whether fan-out alone is enough when one report task breaks.

In this step, get ready to:
  • Add a batch command that dispatches one task per report.
  • Introduce a deterministic inventory failure.
  • Observe a three-report batch with mixed outcomes.
Add the batch command

A group starts one task signature for each requested report. Its GroupResult keeps those child results together for collection.

  • In cli.py, replace the imports and USAGE block at the top with this code:
import sys

from celery import group

from reports import AVAILABLE_REPORTS, write_report
from tasks import app, generate_report

USAGE = """Usage:
  python cli.py sync <report>
  python cli.py submit <report>
  python cli.py status <task-id>
  python cli.py batch <report> [<report> ...]

Available reports: orders inventory customers
"""

What Does This Code Do?

  • The group import makes Celery's group constructor available to the CLI.
  • The new USAGE line documents a batch command that accepts one or more report names.
  • Save cli.py.
  • Return to the first activated Terminal.
  • Reveal the updated usage text by running this command:
python cli.py

What Should I See?

You should see the usage guide with python cli.py batch <report> [<report> ...] listed beneath the existing commands. This confirms the CLI now advertises the batch path.

  • In cli.py, add this function below show_status():
def run_batch(report_names: list[str]) -> None:
    job = group(generate_report.s(name) for name in report_names)
    batch_result = job.apply_async()
    print(f"Batch ID: {batch_result.id}")
    values = batch_result.get(timeout=20, propagate=False)
    for name, value in zip(report_names, values):
        print(f"{name}: {value!r}")

How Does the Batch Work?

  • The group(...) call creates one generate_report signature for each report name.
  • The .apply_async() call dispatches the complete group to the worker.
  • The .get(timeout=20, propagate=False) call collects every child outcome in its original position.
  • The zip(...) loop pairs each report name with its returned value.
  • In cli.py, find this final branch inside main():
    elif command == "status" and len(arguments) == 1:
        show_status(arguments[0])
    else:
        raise SystemExit(USAGE)

What Are You Locating?

This branch currently handles the status command before sending every other command to the usage message. The batch branch belongs immediately before that fallback.

  • Replace that branch with this expanded version:
    elif command == "status" and len(arguments) == 1:
        show_status(arguments[0])
    elif command == "batch" and arguments:
        run_batch([validate_report(name) for name in arguments])
    else:
        raise SystemExit(USAGE)

How Is the Command Connected?

The new branch validates every supplied report name before passing the list to run_batch(). An empty batch still falls through to the usage message.

  • Save cli.py.
  • Test the group path with one report by running this command:
python cli.py batch orders

What Should I See?

You should see a batch ID followed by orders: 'output/orders.csv'. Good progress. Your CLI now dispatches a group and collects its result.

Batch Not Completing?

  • Check that the Celery worker from the previous step is still running in the second Terminal.
  • Check that the first Terminal still shows the activated .venv environment.
  • Confirm that your local Redis service remains available.

Still stuck? Help me debug why my Celery group does not return its orders result.

✔️ Awesome, I've got everything!

Your cli.py now validates report names, dispatches a group, and prints each child result.

ⓧ I'd like to double check the full code

  • Compare your complete cli.py with this reference:
import sys

from celery import group

from reports import AVAILABLE_REPORTS, write_report
from tasks import app, generate_report

USAGE = """Usage:
  python cli.py sync <report>
  python cli.py submit <report>
  python cli.py status <task-id>
  python cli.py batch <report> [<report> ...]

Available reports: orders inventory customers
"""


def validate_report(report_name: str) -> str:
    if report_name not in AVAILABLE_REPORTS:
        raise SystemExit(f"Unknown report: {report_name}\n\n{USAGE}")
    return report_name


def run_sync(report_name: str) -> None:
    output_path = write_report(report_name)
    print(f"Created {output_path}")


def submit_report(report_name: str) -> None:
    result = generate_report.delay(report_name)
    print(f"Task ID: {result.id}")


def show_status(task_id: str) -> None:
    result = app.AsyncResult(task_id)
    print(f"State: {result.state}")
    if result.ready():
        value = result.get(propagate=False)
        print(f"Result: {value!r}")


def run_batch(report_names: list[str]) -> None:
    job = group(generate_report.s(name) for name in report_names)
    batch_result = job.apply_async()
    print(f"Batch ID: {batch_result.id}")
    values = batch_result.get(timeout=20, propagate=False)
    for name, value in zip(report_names, values):
        print(f"{name}: {value!r}")


def main() -> None:
    if len(sys.argv) < 2:
        raise SystemExit(USAGE)

    command = sys.argv[1]
    arguments = sys.argv[2:]

    if command == "sync" and len(arguments) == 1:
        run_sync(validate_report(arguments[0]))
    elif command == "submit" and len(arguments) == 1:
        submit_report(validate_report(arguments[0]))
    elif command == "status" and len(arguments) == 1:
        show_status(arguments[0])
    elif command == "batch" and arguments:
        run_batch([validate_report(name) for name in arguments])
    else:
        raise SystemExit(USAGE)


if __name__ == "__main__":
    main()

This reference contains the complete batch-enabled CLI for this step. The command routing at the bottom should match exactly.

Introduce the inventory failure

The group path now works. A deterministic exception gives you a reproducible way to observe what the group does when one child task breaks.

This temporary version raises the same inventory error on every attempt. The other report tasks keep their existing behavior.

  • In tasks.py, find the current generate_report() task:
@app.task
def generate_report(report_name: str) -> str:
    return write_report(report_name)

What Are You Replacing?

The current task sends every valid report name straight to write_report(). Every report therefore follows the same successful path.

  • Replace the current task with this temporary failure version:
@app.task
def generate_report(report_name: str) -> str:
    if report_name == "inventory":
        raise ConnectionError("Simulated temporary inventory service outage")
    return write_report(report_name)

How Is the Failure Controlled?

  • The condition targets only the inventory report.
  • The raised ConnectionError stops that task before it can write a CSV file.
  • The final return keeps orders and customers on the successful report path.
  • Save tasks.py.

✔️ Awesome, I've got everything!

Your generate_report() task now creates a controlled inventory exception while preserving the other report paths.

ⓧ I'd like to double check the full code

  • Compare your complete tasks.py with this reference:
from celery import Celery

from reports import write_report

app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/0",
)


@app.task
def generate_report(report_name: str) -> str:
    if report_name == "inventory":
        raise ConnectionError("Simulated temporary inventory service outage")
    return write_report(report_name)

This reference is the complete temporary task file for the designed failure. Only inventory raises the simulated exception.

Run the mixed-outcome batch

The running worker still has the earlier task definition in memory. Restarting it loads the controlled inventory failure from the saved tasks.py file.

  • Switch back to the second activated Terminal that runs the worker.
  • Press Ctrl+C to stop the current worker.
  • Restart the worker with the updated task definition by running:
celery -A tasks worker --loglevel=INFO

Why Restart the Worker?

The restarted worker imports the latest tasks.py and registers tasks.generate_report again. Keep this Terminal running so it can receive the batch.

Before you run the batch, which report tasks do you think will return paths? Which task do you think will record an exception?

  • Switch back to the first activated Terminal.
  • Submit the three-report batch by running this command:
python cli.py batch orders inventory customers

What Should I See?

  • The orders entry contains 'output/orders.csv'.
  • The inventory entry contains a ConnectionError with the simulated outage message.
  • The customers entry contains 'output/customers.csv'.
  • The worker logs the inventory task as FAILURE.

That mixed result is intentional. The group preserves the successful reports while exposing the failed child task.

Not Seeing the Mixed Result?

  • If inventory succeeds, confirm that you saved tasks.py before restarting the worker.
  • If every entry fails, confirm that the worker started from the celery-report-pipeline folder.
  • If the batch cannot connect, confirm that your existing local Redis service is still running.

Need another set of eyes? Help me diagnose why my Celery batch does not show two CSV paths and one inventory failure.

You now have a parallel batch that makes one unrecovered failure visible. Next, you will teach the inventory task to recover from that temporary outage.

Recover with a Bound Task Retry

The last batch proved that parallel work can still finish with a missing result. Orders and customers produced CSV files while inventory ended in FAILURE.

This step gives inventory one controlled retry through Celery's bound task pattern. The task reads its retry count before deciding whether to simulate the outage.

In this step, get ready to:
  • Convert generate_report into a bound task with one retry.
  • Requeue the first inventory failure after one second.
  • Rerun the batch to confirm all three reports succeed.
Replace the failure with a bound retry

A bound task receives its Celery task instance as self. That instance exposes the current retry count through self.request.retries.

The retry limit gives the simulated outage one recovery attempt. A second failure would end the task with FAILURE.

  • Switch back to tasks.py in Visual Studio Code.
  • Select the existing @app.task decorator through return write_report(report_name).
  • Replace the selected function with the code below:
@app.task(bind=True, max_retries=1)
def generate_report(self, report_name: str) -> str:
    if report_name == "inventory" and self.request.retries == 0:
        try:
            raise ConnectionError("Simulated temporary inventory service outage")
        except ConnectionError as exc:
            raise self.retry(exc=exc, countdown=1)

    return write_report(report_name)

How does retry recovery work?

  • The bind=True setting passes the task instance into the function as self.
  • The max_retries=1 setting allows one retry after the initial attempt.
  • The self.request.retries == 0 condition limits the simulated outage to the first inventory attempt.
  • The raise self.retry(exc=exc, countdown=1) line requeues the same task after one second.
  • Celery stores the intermediate state as RETRY in the Redis result backend.
  • The retry count becomes 1 on the second attempt. The task reaches write_report(report_name) and creates the CSV file.
  • Save tasks.py.

✔️ Awesome, I've got everything!

Great. Your saved tasks.py now gives the inventory job one controlled recovery attempt.

ⓧ I'd like to double check the full code

  • Compare your saved tasks.py with the full file below:
from celery import Celery

from reports import write_report

app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/0",
)


@app.task(bind=True, max_retries=1)
def generate_report(self, report_name: str) -> str:
    if report_name == "inventory" and self.request.retries == 0:
        try:
            raise ConnectionError("Simulated temporary inventory service outage")
        except ConnectionError as exc:
            raise self.retry(exc=exc, countdown=1)

    return write_report(report_name)
Restart the Celery worker

The worker imports tasks.py when its process starts. Restarting the worker loads the bound task definition you just saved.

  • Switch back to the Celery worker terminal from earlier.
  • Stop the existing worker by pressing Control+C.
  • Restart the worker with the bound task by running:
celery -A tasks worker --loglevel=INFO

What does this command do?

  • The -A tasks option loads the Celery app from tasks.py.
  • The worker command starts the process that receives queued reports.
  • The --loglevel=INFO option keeps task state messages visible during the retry.

Your worker should register tasks.generate_report and remain active in this terminal. It is now ready to execute the updated task.

Worker does not start?

  • Compare tasks.py with the full-code tab if the worker reports a Python syntax problem.
  • Confirm the earlier worker process stopped before starting the replacement.
  • Check that the terminal prompt still shows the activated .venv environment if the Celery command is unavailable.

Still stuck? Help me diagnose why my Celery worker cannot load the bound generate_report task.

Rerun the recovered batch

The batch still submits one task for each requested report. The updated worker now gives the first inventory failure a route back into the queue.

Before you rerun the batch, do you expect inventory to stay failed or recover on its second attempt?

  • Switch back to the first activated terminal from earlier.
  • Run the recovered three-report batch by running:
python cli.py batch orders inventory customers

What happens during this run?

  • The Celery Canvas group queues one task for each report.
  • The first inventory attempt raises the simulated connection error.
  • The retry places that inventory task back on the queue for one second.
  • The second inventory attempt calls write_report(report_name).
  • The GroupResult waits for every report result before the CLI prints the paths.

Your CLI prints one path for each report. The paths include output/orders.csv, output/inventory.csv, and output/customers.csv.

The worker log shows RETRY for inventory. A later line shows SUCCESS for the same task.

  • Return to Visual Studio Code.
  • Select output/inventory.csv in the Explorer sidebar.

You'll see items_in_stock,128 on one row. You'll see items_low_stock,7 on the next row.

Inventory still failing?

  • Confirm you saved tasks.py before restarting the worker.
  • Confirm the running worker started after the code change.
  • Keep the worker terminal active while the batch command waits for its results.

Need another pair of eyes? Help me troubleshoot why inventory does not move from RETRY to SUCCESS.

Strong work. The inventory report now survives its simulated outage. Your batch finishes with all three CSV paths.

Secret mission

Summarize the Batch with a Chord

Turn three independent report results into one durable summary. You will add a Celery chord that waits for retry recovery before writing `output/batch_summary.json`.

Clean Up Your Resources

Clean Up Your Resources

All resources stay on your Mac, so this project has no ongoing service charges. Choose whether to keep the pipeline running, pause its processes, or remove its local files.

Resources you used:

  • A foreground Celery worker process that runs report tasks.
  • Project source files inside celery-report-pipeline.
  • The .venv virtual environment inside celery-report-pipeline.
  • Generated report artifacts inside output.

Protect Shared Redis Data

The project stored task results in database 0 of your existing Redis Open Source service. Database 0 may also contain records from other local work.

This cleanup leaves database 0 unchanged. Your existing Redis installation remains available for other projects.

Keep everything running

No action is needed. Choose this option if you want to keep submitting reports or experimenting with the workflow.

  • Leave the Celery worker running in its terminal.
  • Keep the existing Redis service active.
  • Keep the celery-report-pipeline folder with its virtual environment and generated reports.

Pause - I'll come back to this later

Stop the background process to free up local resources. Your project files remain ready for another session.

  • Return to the terminal running the Celery worker.
  • Press Ctrl+C to stop the worker gracefully.

That frees the worker's resources while your source files and generated reports stay available.

  • Leave Redis running if another local project uses it.
  • Stop the existing Redis service only if it was running solely for this project by running:
brew services stop redis

What Does This Command Do?

Homebrew stops the existing Redis service. The Redis installation and its local data remain on disk.

  • Confirm that Redis is paused by running this health check. A connection failure is the expected result:
redis-cli ping

What Should I See?

This health check should fail to return PONG because Redis is paused. The exact connection message can vary.

Still Seeing PONG?

A PONG response means Redis is still running. Help me check why my Redis service did not stop..

Delete - I don't want to use this again

Removing celery-report-pipeline permanently removes your source files, virtual environment, CSV reports, and JSON summary. Your existing Redis installation stays available for other local work.

  • Return to the terminal running the Celery worker.
  • Press Ctrl+C to stop the worker gracefully.
  • Click Finder in the Dock.
  • Locate the celery-report-pipeline folder where you created it.
  • Move the celery-report-pipeline folder to your Mac's Trash.
  • Empty the Trash when you are ready to remove the project files permanently.
  • Keep the pre-existing Redis installation in place.
  • Keep Redis database 0 unchanged because it may contain unrelated local data.

Your project-owned files and worker process are now removed. The shared Redis service remains protected.

Nice Work!

Nice Work!

You did it: you built a resilient local report pipeline with Celery. Redis carries the task data while your worker completes parallel reports with retry recovery.

You've learned how to:

  • Built a CSV generator with synchronous processing. Its visible three-second pause exposed why report work blocks the foreground process.
  • Moved report generation to a background Celery worker behind a Redis broker. The CLI returns an immediate task ID. Its status command reads the task state plus completed CSV path from the result backend.
  • Fanned three reports out with a Celery group. You observed the inventory failure. You recovered it with a bound task retry.
  • Completed the optional Secret Mission by coordinating the batch with a Celery chord. Its callback writes one JSON summary to output/batch_summary.json after every report succeeds.

Ready to quiz yourself?