Tejas Shah
03/13/2025, 11:24 AMfrom prefect import flow, serve, allow_failure
@flow(log_prints=True, flow_run_name="child_flow_{flow_param}")
def child_flow(flow_param: int):
print(f"In flow {flow_param}")
if flow_param==5:
raise ValueError("child_flow_5 always fails")
@flow(log_prints=True)
def parent_flow():
for i in range(0,11):
allow_failure
if __name__ == "__main__":
parent_flow_deployment = parent_flow.to_deployment(name="parent_flow_dev_deployment")
serve(parent_flow_deployment)Marvin
03/13/2025, 11:24 AMparent_flow continues executing even if child_flow_5 fails, you should use the allow_failure function correctly. The allow_failure decorator allows a task or flow to fail without failing the entire flow run.
Here's a corrected version of your code:
python
from prefect import flow, serve, allow_failure
@flow(log_prints=True, flow_run_name="child_flow_{flow_param}")
def child_flow(flow_param: int):
print(f"In flow {flow_param}")
if flow_param == 5:
raise ValueError("child_flow_5 always fails")
@flow(log_prints=True)
def parent_flow():
for i in range(0, 11):
allow_failure(child_flow)(flow_param=i)
if __name__ == "__main__":
parent_flow_deployment = parent_flow.to_deployment(name="parent_flow_dev_deployment")
serve(parent_flow_deployment)
### Key Changes:
- Use allow_failure: Wrap the child_flow call with allow_failure to let it fail without stopping the parent_flow.
### Resources:
- Prefect 3.x API Reference
- Prefect 3.x Documentation
This should allow your parent_flow to continue executing all child flows, even if child_flow_5 raises an exception. If you have any more questions or need further assistance, feel free to ask.Tejas Shah
03/13/2025, 11:26 AM'allow_failure' object is not callableMarvin
03/13/2025, 11:26 AMallow_failure. Let's correct that.
To allow a child flow to fail without stopping the parent flow, you can use the try-except block directly within your flow. Here's how you can modify your parent_flow to handle failures in `child_flow_5`:
python
from prefect import flow, serve
@flow(log_prints=True, flow_run_name="child_flow_{flow_param}")
def child_flow(flow_param: int):
print(f"In flow {flow_param}")
if flow_param == 5:
raise ValueError("child_flow_5 always fails")
@flow(log_prints=True)
def parent_flow():
for i in range(0, 11):
try:
child_flow(flow_param=i)
except Exception as e:
print(f"Child flow {i} failed with exception: {e}")
if __name__ == "__main__":
parent_flow_deployment = parent_flow.to_deployment(name="parent_flow_dev_deployment")
serve(parent_flow_deployment)
### Key Changes:
- Wrap child_flow call in a try-except block: This allows each child flow to fail independently without affecting the rest of the flow execution.
This approach should allow your parent_flow to continue executing all child flows, even if one of them raises an exception. If you have further questions, feel free to ask.