Great question — you can do this cleanly in Prefect 2.x by submitting each item as its own task, inspecting each task’s state, only chaining the “next” task for successful items, and then sending a summary report at the end. Below is a complete pattern you can copy/paste. It: - Runs per-item work in parallel - Continues processing even if some items fail - Triggers a follow-up task only for successful items - Builds a success/failure report - Sends the report to Slack via a Slack Webhook block Links to docs: - Tasks and flows: Tasks | Flows - States and results: States | Results - Slack integration: Slack credentials and Slack messages - Automations (Cloud): Automations Example flow

from prefect import flow, task, get_run_logger
from prefect_slack import SlackWebhook

# Task that might fail
@task(retries=2, retry_delay_seconds=5)
def process_item(item: dict) -> dict:
    # Do your real work here
    if item.get("should_fail"):
        raise ValueError(f"Processing failed for {item['id']}")
    return {"id": item["id"], "value": item["payload"] * 2}

# Task to run only if process_item succeeds
@task
def after_success(item_id: str, result: dict):
    # e.g., write to a DB, publish to a topic, etc.
    return f"follow-up-complete-for-{item_id}"

def build_report(results):
    succeeded = [r for r in results if r["status"] == "succeeded"]
    failed = [r for r in results if r["status"] == "failed"]
    lines = []
    lines.append(f"Total: {len(results)} | Succeeded: {len(succeeded)} | Failed: {len(failed)}")
    if failed:
        lines.append("")
        lines.append("*Failures:*")
        # show a few failures for readability
        for r in failed[:10]:
            lines.append(f"- item={r['item_id']} error={r['error']}")
        if len(failed) > 10:
            lines.append(f"...and {len(failed) - 10} more")
    return "\n".join(lines)

def notify_slack(block_name: str, subject: str, body: str):
    webhook = SlackWebhook.load(block_name)
    webhook.notify(body, subject=subject)

@flow(name="per-item-exception-handling")
def orchestrate(items: list[dict], slack_block: str = "my-slack-webhook"):
    """
    items: list like [{"id": "a1", "payload": 3, "should_fail": False}, ...]
    slack_block: name of a Prefect Slack Webhook block
    """
    logger = get_run_logger()

    # Submit per-item tasks concurrently
    futures = {item["id"]: process_item.submit(item) for item in items}

    # Inspect each result; only chain next task for successes
    results = []
    followup_futures = {}
    for item_id, fut in futures.items():
        state = fut.wait()  # wait for each to finish; does NOT raise
        if state.is_failed():
            err = state.result(raise_on_failure=False)  # returns the exception object
            results.append({"item_id": item_id, "status": "failed", "error": repr(err)})
        else:
            value = state.result()  # safe to get the value on success
            followup_futures[item_id] = after_success.submit(item_id, value)
            results.append({"item_id": item_id, "status": "succeeded"})

    # Ensure follow-up tasks finish
    for item_id, fut in followup_futures.items():
        fut.wait()

    # Build and send Slack report
    report = build_report(results)
    <http://logger.info|logger.info>("\n" + report)
    notify_slack(slack_block, subject="Prefect per-item report", body=report)

    return results

How to set up the Slack webhook block - In Prefect Cloud UI: Blocks > Create > Slack Webhook, paste your webhook URL, and name it (e.g.,

my-slack-webhook

). - Or via code once: ```

from prefect_slack import SlackWebhook SlackWebhook(url="https://hooks.slack.com/services/XXX/YYY/ZZZ").save( "my-slack-webhook", overwrite=True )

Alternative pattern: catch exceptions inside the task
If you prefer to treat “errors” as data (so the task never fails), wrap the logic in try/except and return a status dict. This makes downstream branching simpler, and the flow won’t have any failed task states (useful if you want a green flow even with per-item errors).

@task def process_item_safe(item): try: # real work if item.get("should_fail"): raise ValueError("boom") return {"ok": True, "id": item["id"], "value": item["payload"] * 2} except Exception as e: return {"ok": False, "id": item["id"], "error": str(e)} @flow def orchestrate_safe(items: list[dict], slack_block: str = "my-slack-webhook"): results = [process_item_safe.submit(i) for i in items] results = [r.result() for r in results] successes = [r for r in results if r["ok"]] failures = [r for r in results if not r["ok"]] # Only chain for successes for r in successes: after_success.submit(r["id"], r) report = f"Total: {len(results)} | Succeeded: {len(successes)} | Failed: {len(failures)}" SlackWebhook.load(slack_block).notify(report, subject="Prefect per-item report") ``` Optional: use Prefect Cloud Automations Since you’re on Prefect Cloud, you can also set up an Automation to notify Slack when the flow completes or fails globally (no code changes). This is great as a “meta” alert; you can keep the per-item report in-code if you want details. - Docs: Automations - Create an Automation that triggers on

prefect.flow-run.Completed

(or

Failed

) and sends a Slack notification using your Slack block. Would you like me to tailor this for your exact notification channel (Slack vs email), or adapt it to run on your GCP runtime (e.g., Cloud Run or GKE)?