e

    Ethan Shenker

    2 years ago
    Hi everyone! I'm working on an internship project over this summer and have a question about working with Prefect states. Part of the goal of my workflow is to allow the user to manually trigger a task once a previous task has "failed". I plan on achieving this by implementing the
    signals.FAIL()
    state, which would then trigger another task. However, within that downstream task, I would also like to allow the user to interact with the flow to set a flag that would determine the outcome of that downstream flow, and this interaction would likely occur while the downstream task was in a paused state. However, I'm struggling with determining how I'd be able to achieve this interaction, as the paused state pauses all elements of the flow, thus preventing the user interaction I require. Additionally, if there are any other ways that a task can be set to occur both as a result of a failed previous task and manually at the same time, that insight would be greatly appreciated too. Thanks!
    Jim Crist-Harif

    Jim Crist-Harif

    2 years ago
    Hi Ethan, can you expand on your use case here a bit? To simplify, say this is a flow with two tasks A and B, where A is upstream of B. • Do you always want task B to run manually, or only if task A fails? • What kind information do you want the user to manually convey to task B to determine the outcome? • Are you running with an orchestration layer (cloud or server), or just using prefect core (e.g.
    flow.run()
    )? • If you are using orchestration, do you want the manual trigger to happen using the prefect UI, or some other interaction?
    e

    Ethan Shenker

    2 years ago
    Hi Jim, thank you for reaching out. To respond to each point individually: • I want task B to only run if task A fails. If task A were to succeed, I'd want task B to be completely forgotten about. • The information being conveyed to task B is effectively a boolean flag, (Do you want this task to overwrite information that already exists in a database with information produced by task A). If the user says yes, then the task would continue its run, but it would exit if the user responded "false" (or whatever negative flag was chosen) • I believe production-time running is being done by the server, but I would also like to be able to test using the core. • Using the orchestration, I'd like the manual trigger to occur using the UI. Again, thank you so much! Hope this helps!
    Jim Crist-Harif

    Jim Crist-Harif

    2 years ago
    You could do this with a custom trigger for task B. • If task A fails, you'd raise
    prefect.engine.signals.PAUSE
    in the trigger, which would put task B into a manual-only run mode. • If task A is in any other state, you'd raise
    prefect.engine.signals.SKIP
    to skip task B. Once task B is paused, the user could use the UI to either resume task B (this triggering it to run and overwrite) or transition it to finished directly (the task won't run, no overwriting would happen). I believe this satisfies your workflow?
    The documentation on triggers is a bit sparse, you might look here: https://docs.prefect.io/core/concepts/execution.html#triggers and then at the code https://github.com/PrefectHQ/prefect/blob/master/src/prefect/triggers.py for examples.
    e

    Ethan Shenker

    2 years ago
    So rather than taking in an input, the workaround would be to either continue running the task or just to skip it altogether?
    Jim Crist-Harif

    Jim Crist-Harif

    2 years ago
    That's what I'd suggest, yeah.
    We don't currently provide a way for a user to provide additional context when resuming a task (although we could support this). Since it sounds like you want to use that input to either skip or run the task, it seems simpler to do that directly rather than in the task logic.
    e

    Ethan Shenker

    2 years ago
    Got it. That seems to make much more sense than what I had originally thought. Just as a follow up, would I define that task as:
    @task(name="task b", trigger=triggers.any_failed)
    def task_b():
      raise signals.PAUSE()
      [rest of code]
    or would the trigger be something else?
    Jim Crist-Harif

    Jim Crist-Harif

    2 years ago
    That could work. I was suggesting you write a custom trigger, but your suggestion above seems simpler.
    e

    Ethan Shenker

    2 years ago
    Especially because the notion of writing my own trigger for this task specifically is a little beyond my current scope of Prefect knowledge😅
    Again, thank you so much for the help! It's beyond appreciated 😃
    Jim Crist-Harif

    Jim Crist-Harif

    2 years ago
    Glad to help! Hope it all goes well, please feel free to reach out if you run into any more issues.
    e

    Ethan Shenker

    2 years ago
    I definitely will! Thank you!
    Hi @Jim Crist-Harif! Another question has come up over the course of the past day. I got it so that the task pauses after having been triggered by a any_failed trigger, but now I'm having trouble unpausing a task from the UI.