i am dynamically scheduling a flow run (to start n...
# prefect-community
s
i am dynamically scheduling a flow run (to start not immediately but in the future) from within a flow and am polling the prefect AI to wait for its completion
Copy code
def get_flow_run_state(client, flow_run_id):
    q = parse_graphql(
            {'query': {
                with_args('flow_run', 
                    {'where': {'id': {'_eq': flow_run_id}}}): {
                        'state'
                        }
                    }
                    })
    state = client.graphql(q).data.flow_run[0].state
    return state

def wait_flow_complete(flow_run_id):
    client = Client()

    state = None
    while state != 'Success':
        sleep(10)
        state = get_flow_run_state(client, flow_run_id)
question: is it possible to avoid the polling and achieve the same thing?
j
You might want to use the FlowRun Task to start a flow within a flow https://docs.prefect.io/api/latest/tasks/prefect.html#flowruntask with kwarg
wait=True
downstream tasks will wait for the completed FlowRun, although, this terminal state must not be a success state.
upvote 1
n
Thanks for the feedback. I’ll look into it 👍
Took the chance to look more closely to the documentation… I’m not sure how this would help me though….
s
@Julian i decided i couldn’t use
FlowRunTask
because i didn’t want the task to start immediately but want to schedule it for later (while still having upstream and downstream dependencies for the task)
• schedule the task only after an upstream task is done • start a downstream task only after that scheduled task has later completed