@Marvin Can you give me an example of handling exceptions out of a mapped task?

Certainly. Handling exceptions in mapped tasks in Prefect is similar to handling exceptions in regular tasks, but you need to be aware that each instance of a mapped task runs independently. So, you'll have to consider exception handling within each mapped task's execution context. Here's a conceptual example using Python code:

python
from prefect import Flow, task
from prefect.engine.signals import SKIP

@task
def process_item(item):
    try:
        # Your processing logic here
        result = item / 0  # This will cause a ZeroDivisionError
        return result
    except ZeroDivisionError:
        # Handle the exception by perhaps skipping the task
        raise SKIP('Divided by zero. Skipping this item.')
    except Exception as e:
        # Handle any other exceptions that may occur
        print(f"An error occurred: {e}")
        raise

@task
def prepare_data():
    # Prepare your data list here
    return [1, 2, 3, 4, 5]

with Flow("Exception Handling in Mapped Tasks") as flow:
    data = prepare_data()
    results = process_item.map(data)

# Running the flow
flow_state = flow.run()

In this example, if

process_item

encounters a

ZeroDivisionError

, it will use the

SKIP

signal to skip that particular mapped task instance. Other exceptions can be handled in a different catch block, where you can log the error, raise it, or handle it as needed. When dealing with exceptions in mapped tasks, you can tailor the behavior depending on what's appropriate for your application. You might want to skip the task, fail it, retry it, or record the error and continue with other processing.

🙅 use

.map(..., return_state=True)

and then you can check if the state is completed/failed and/or fetch the result

Ahh thanks Nate. It had slipped my mind regarding the state return. Makes sense!