Hi, I would like to get flow_run_id from a flow created via create_flow_run(), how would I be able to do that? For context, I’m trying to create a FlowRunView of a child flow created by create_flow_run().

create_flow_run

returns the flow run id already

I tried to create FlowRunView by doing:

child_flow_run_id = create_flow_run(
        flow_name="test-produce-logs-flow", #, run_name="custom_run_name"
        project_name="data_quality_tracking"
    )

    flow_run_view = FlowRunView.from_flow_run_id(
        child_flow_run_id
    )

But I’m getting an error that says

GRAPHQL_PARSE_FAILED: Syntax Error: Cannot parse the unexpected character "<".

Is there something I have to do to

child_flow_run_id

before I can put it into

FlowRunView.from_flow_run_id()

?

but this is conflating build time and runtime because while the DAG is building, the

child_flow_run_id

is of type task but the

FlowRunView.from_flow_run_id

is not a task so it executes immediately. So it’s either you run the create_flow_run eagerly outside of a Flow or you put the FlowRunView logic inside a task

Ah I keep conflating build time and runtime! Sorry you have to keep telling me this every week 😅 Getting a hang of it!

it’s less obvious on this one haha

A follow up question, if I wanted to put a

wait_for_flow_to_run

AND create a

FlowRunView

, how would I be able to do that? I was able to find examples of both used separately, but no examples for them being used together

You can just keep the id right?

with Flow() as flow:
    the_id = create_flow_run()
    wait_for_flow_run(the_id)
    task_with_flow_run_view(the_id)

or you can make a new task that calls

wait_for_flow_run.run()

and then adds some logic

When I try the above, it seems like the

task_with_flow_run_view

isn’t actually waiting for

wait_for_flow_run

to be done. Should the below schematic be linear as well if I implemented the

wait_for_flow_run

correctly?

Wait just

wait_for_flow_run

allows your to stream logs if that is your goal

Am I able to get the logs as an object?

Ahh I see what you mean. You can specify the upstream task right?

with Flow(...) as flow:
    get_logs.set_upstream(wait_for_flow_run_task)

Yep that worked! Thank you!