@Marvin I would like to do the follow: I would like to set a flow to Failed but not terminate the flow at the same time. Context: I am fetching an API day by day for daily data, and I would like to indicate in the UI that for one day I got and error or no data. Currently I just log this using a try catch block, but I would like to have an indicator that actually this was a failed flow, where there are some missing data. example code below. How can I do this in Prefect v3?

df = pl.DataFrame()
            try:
                df = fetch_data(
                    start_date=load_from,
                    end_date=load_to,
                )

            except Exception as e:
                logger.error(
                    f"Error fetching data for {in_domain} <-> {out_domain}: {e}"
                )
                continue

. ```

✅ thought for 336.0 seconds

I don't want to stop the whole flow, but I would like to set the flow state to failed if there is missing data/exception

Short answer: you can’t “fail a flow and keep it running” in a supported way. In Prefect 3.x the engine determines the final state at the end of execution. You can hack the state mid-run via the client, but it’s confusing and not recommended. What you probably want is: - Keep processing each day (catch exceptions so the loop continues) - Surface which days failed in the UI (artifacts and/or child runs) - Mark the parent flow as Failed at the end if any day failed Two good patterns 1) Single flow that aggregates errors, then fails at the end - Keep your try/except and continue - Track failures - Optionally create an artifact summarizing failed days - At the end, raise an exception to mark the flow as Failed Example:

from datetime import date, timedelta
from prefect import flow, get_run_logger
from prefect.artifacts import create_table_artifact

def date_range(start: date, end: date):
    d = start
    while d <= end:
        yield d
        d += timedelta(days=1)

def fetch_data(start_date, end_date):
    # your existing function
    ...

@flow
def fetch_daily_range(load_from: date, load_to: date):
    logger = get_run_logger()

    failed = []
    collected = []

    for day in date_range(load_from, load_to):
        try:
            df = fetch_data(start_date=day, end_date=day)
            collected.append(df)
        except Exception as e:
            logger.error(f"Error fetching data for {day}: {e}")
            failed.append({"date": day.isoformat(), "error": str(e)})
            continue

    if failed:
        # Optional: show a table in the UI with the failures
        create_table_artifact(
            table=failed,
            description=f"{len(failed)} day(s) failed during fetch"
        )
        # Fail the flow AFTER finishing all days
        raise RuntimeError(f"Partial failure: {len(failed)} day(s) failed")

    return collected

2) Subflow per day (best for clear UI visibility and retries) - Wrap “fetch one day” in a child flow so the UI shows per-day runs - Parent catches failures so it continues - Optionally fail parent at the end if any child failed Example:

from prefect import flow, get_run_logger
from prefect.artifacts import create_table_artifact

@flow(name="fetch_one_day")
def fetch_one_day(day):
    # Raises if it fails (child run will show Failed)
    return fetch_data(start_date=day, end_date=day)

@flow(name="fetch_daily_range")
def fetch_daily_range(load_from, load_to):
    logger = get_run_logger()
    failed = []

    for day in date_range(load_from, load_to):
        try:
            fetch_one_day(day)  # each day is its own flow run in the UI
        except Exception as e:
            logger.error(f"Error fetching data for {day}: {e}")
            failed.append({"date": day.isoformat(), "error": str(e)})
            continue

    if failed:
        create_table_artifact(
            table=failed,
            description=f"{len(failed)} day(s) failed"
        )
        raise RuntimeError(f"Partial failure: {len(failed)} day(s) failed")

Why not set the flow to Failed mid-run? - You can technically call the orchestration client and set a Failed state mid-run, but the engine will continue executing and then try to set a final state at the end. This leads to confusing state transitions and can break automations/triggers downstream. If you really need a mid-run state update (not recommended) - You could do something like:

from prefect.client import get_client
from prefect.states import Failed
from prefect.runtime import flow_run

client = get_client(sync_client=True)
client.set_flow_run_state(
    flow_run_id=flow_run.id,
    state=Failed(message="Partial failure encountered so far"),
    force=True,
)

…but again, this is not the intended pattern. Nice-to-have UI indicators - Create a table or markdown artifact summarizing which dates failed: ``` from prefect.artifacts import create_table_artifact, create_markdown_artifact