https://prefect.io logo
Join Slack
Powered by
Hi all, I'm currently performing some testing on t...
# prefect-community
n
Hi all, I'm currently performing some testing on the self-hosted version of prefect. I have a flow which contains a task that needs to be resumed manually in order to continue the execution of the flow. On the cloud version, it worked perfectly but on the self-hosted one, I could not find any way to resume this task. That means my flow remains in the running state. Does anyone know how to do that? Thanks in advance!
j
Hi @Nicolas van de Walle! That UI feature will be coming to the open source product in a release soon. In the meantime, if you manually put the task into a RESUME state (from the UI or via GraphQL), it should achieve the same.
n
Well, thank you for your reply. I could not find how to put that specific task in a certain state. The "mark as" function of the UI says I "can only change the marked state of a finished task". Otherwise, do you have any sample GraphQL query that would perform that operation? To be honest, I am new to GraphQL and I did not manage to use this documentation to achieve this task: https://docs.prefect.io/orchestration/concepts/flow_runs.html#graphql-2
j
Ah, we put that guardrail in to prevent unexpected behavior from changing in-progress tasks. One sec I’ll give you the GQL you need
If you want to do this entirely from the Python client:
Copy code
c = prefect.Client()
c.set_task_run_state(
    <YOUR TASK RUN ID>, 
    version=2, 
    state=prefect.engine.state.Resume()
)
A quick way to get the task run ID is by looking at the URL of the UI (I realize this is suboptimal; it has been improved in the unreleased version)
And version=2 will be typical for most Paused tasks; if your task has gone through more states, it may be higher.
Here’s how you can query for this information:
Copy code
c.graphql(
    """
    query {
        task_run(where: {id: {_eq: <TASK_RUN_ID>}}) {
            state
            version
        }
    }
    """
)
n
Thanks a lot, I will look into it !
Is there a way of performing the following request using the GraphQL API?
Copy code
c = prefect.Client()
c.set_task_run_state(
    <YOUR TASK RUN ID>, 
    version=2, 
    state=prefect.engine.state.Resume()
)
I looked at the setTaskRunState mutation which is deprecated so I tried set_task_run_state*s* but I could not understand how this one works. Here is what I've got so far:
Copy code
mutation {
  set_task_run_states(input: {
    states: {
      task_run_id: "TASK_RUN_ID"
      version: 2
      state: "{\"type\":\"Success\"}"
    }
  }) {
    states {
      id
      status
      message
    }
  }
}
This works well, it changes the task state but does not go further. The flow does not continue after this change (and does not end either if it was the last task). @Jeremiah do you have an idea of what I might be doing wrong? Thank you!
j
Very close - it is expecting a list of states, like this:
Copy code
mutation {
  set_task_run_states(input: {
    states: [ # <- note this bracket
    {
      task_run_id: "TASK_RUN_ID"
      version: 2
      state: "{\"type\":\"Success\"}"
    }
    ]  # <- note this bracket
  }) {
    states {
      id
      status
      message
    }
  }
}
n
Thanks !!
Open in Slack
PreviousNext
Powered by