https://prefect.io logo
Channels
Powered by
m

Mathijs Schoorl

03/05/2021, 2:12 PM
I'm playing around with the trigger manual_only. Which seems to work great. Just something I can not really find a good answer to: Is it possible to only ask for an approval if all upstream tasks are successful? It looks like manual_only means it will always ask for approval and keeps the flow running. I would like to have an approval task fail with the status TRIGGERFAIL if any upstream tasks are in a failed state.
n

nicholas

03/05/2021, 4:03 PM
Hm - I think your best bet here might be to add a task upstream to your manual task that will only continue on 
all_successful
; does that make sense?
m

Mathijs Schoorl

03/05/2021, 4:27 PM
I don't think that will work. As far as I know a task with a trigger 
all_successful
will get the status TRIGGERFAIL when one of his upstream tasks fails. So in the end it will finish with a status and then my Task with trigger 
manual_only
will fire and set to a paused state instead of the TRIGGERFAIL I want.
n

nicholas

03/05/2021, 4:34 PM
You know I think you're right
I think this is worth opening a ticket on the Core repo
Would you mind doing that and describing what your desired behavior would be?
One option in the interim might be to use a 
case
statement to control the instantiation of that task, something like this:
Copy code
control_task = ControlTask(trigger=all_successful)(upstream_tasks=[fail_on_purpose])

with case(control_task, True):
  approve = ApproveTransfer(trigger=manual_only)(upstream_tasks=[control_task])
m

Mathijs Schoorl

03/05/2021, 4:38 PM
I'll give that a try after the weekend. Depending on the outcome I'll create a ticket in the Core repo. Thanks!
n

nicholas

03/05/2021, 4:41 PM
Sounds good - have a nice weekend! 🙂
Ah sorry @Mathijs Schoorl - I gave you incomplete advice, you can also write a custom trigger with the form:
Copy code
trigger_fn(upstream_states: Set[State]) -> bool
m

Mathijs Schoorl

03/08/2021, 7:03 PM
Creating my own trigger worked great, thanks:
Copy code
def manual_trigger_adv(upstream_states: Dict["core.Edge", "state.State"]) -> bool:

    if not all(s.is_successful() for s in _get_all_states_as_set(upstream_states)):
        raise signals.TRIGGERFAIL(
            'Trigger was "manual_only" but some of the upstream tasks failed.'
        )
    if context.get("resume"):
        return True

    raise signals.PAUSE('Trigger function is "manual_only"')
😄 1
2 Views
Powered by