Nils
01/17/2025, 1:12 PMrun_deployment
. How do I make sure the parent flow raises an error if the called deployment failed?Marvin
01/17/2025, 1:13 PMrun_deployment
, you can handle this by checking the state of the deployment run. Here's a concise way to achieve this in Prefect 3.x:
1. Execute the Deployment: Use run_deployment
to execute the child deployment.
2. Check the State: After the deployment run, check its state. If the state is a failure, raise an exception in the parent flow.
Here's a simple example:
python
from prefect import flow, task
from prefect.deployments import run_deployment
from prefect.states import Failed
@flow
def parent_flow():
deployment_run = run_deployment(name="child-deployment-name")
if isinstance(deployment_run.state, Failed):
raise RuntimeError("Child deployment failed")
parent_flow()
### Key Points:
- State Checking: Use isinstance()
to check if the deployment run's state is a Failed
state.
- Error Raising: Raise an exception in the parent flow if the deployment run fails.
For more detailed information, you can refer to the Prefect 3.x documentation.
If you have any more specific questions or need further clarification, feel free to ask!Nils
01/17/2025, 1:18 PMMarvin
01/17/2025, 1:18 PMFailed
is a function, not a type. You should call Failed()
to create a state instance and then compare it to the state of the deployment run. Here's the corrected code:
python
from prefect import flow
from prefect.deployments import run_deployment
from prefect.states import Failed
@flow
def parent_flow():
deployment_run = run_deployment(name="child-deployment-name")
if isinstance(deployment_run.state, Failed()):
raise RuntimeError("Child deployment failed")
parent_flow()
### Key Points:
- State Comparison: Use Failed()
to create an instance for comparison.
- Error Checking: Ensure the state comparison logic is using instances, not the function itself.
If you need further assistance, feel free to ask!