Thread
#prefect-community
    s

    Sanjay Patel

    1 year ago
    Hi, I'm trying to understand how to use StartFlowRun as per Running dependent flows but can't get my head around the concept of passing parameters from one flow into the next. the 'Running a parametrized flow' example here doesn't use a flow as an input. I've done a few searches but can't find exactly what I am looking for. this is one example I was looking at but it doesn't give enough detail
    with Flow('first') as flow:
        a = task_a()
        b = task_b(a)
        c = task_c(b)
        StartFlowRun(flow_name='second', project_name=...)(parameters={"input"=c})
    this is the closest to what I am trying to do but the response ended with using state_result.
    with Flow('first') as flow:
        a = task_a()
        b = task_b(a)
        c = task_c(b)
        
    with Flow('second') as flow:
        param = Parameter('input')
        d = task_d(param)
        
    # How to do something like this:
    flow_a = StartFlowRun(flow_name="first", project_name="examples", wait=True)
    flow_b = StartFlowRun(flow_name="second", project_name="examples", wait=True)
    with Flow('total') as flow:
        a = flow_a()
        b = flow_b()(upstream_tasks=[a], parameters={'input': a.d})
    I have hte option of changing one of my flows so it produces the output I need for the next flow. so 2 questions • how does the output of flow_a get referenced into flow b • how do you access the parameter input that comes from flowa into flowb thank you!
    nicholas

    nicholas

    1 year ago
    Hi @Sanjay Patel - for what you're trying to achieve, think about
    StartFlowRun
    as equivalent to
    flow.run
    or pressing the
    Run
    button in the UI. It boils down to: anything you can do with the latter, you can do with
    StartFlowRun
    , which is a fancy wrapper where the above are instead represented in a task. That task returns something just like any other task, it's just that that task happens to be a
    State
    object. So you could build a flow that orchestrates flows to look something like this:
    @task
    def task_a():
      return "hello"
    
    task_b = StartFlowRun(flow_name="flow_b", project_name="examples", wait=True)
    task_c = StartFlowRun(flow_name="flow_c", project_name="examples", wait=True)
    
    with Flow("flow_a") as flow:
      result_a = task_a()
      result_b = task_b(upstream_tasks=[result_a], parameters={'result_a': result_a})
      result_c = task_c(upstream_tasks=[result_b], parameters={'result_b': result_b.result}) # We access the .result of result_b because StartFlowRun returns a State object, which includes the state message, result, context, and any cached inputs
    Given that
    StartFlowRun
    returns a
    State
    object, any results you want to access from that can be accessed as you would any other data on an object. For more information on the state object, check out the State docs. Hopefully that clears it up a bit!
    s

    Sanjay Patel

    1 year ago
    oh thank you so much. this helps a lot. Let me give this a try and get back to you
    nicholas

    nicholas

    1 year ago
    Happy to help! 😄
    s

    Sanjay Patel

    1 year ago
    Just one follow up please. How would I then include the mapping function as part of this? will task_b.map work the same way where result_a will just be an array instead?
    with Flow("flow_a") as flow:
      result_a = task_a.map(some_iterable)
      result_b = task_b.map(upstream_tasks=[result_a], parameters={'result_a': result_a}) # map
    nicholas

    nicholas

    1 year ago
    @Sanjay Patel - yes, that's correct! And to unclude any parameters that you don't want to map over, you can use the
    unmapped
    utility function.
    j

    Justin Chavez

    1 year ago
    Hi @nicholas, I am trying to perform a similar parameterized flow of flows like Sanjay, except when I run my flow using the result of a StartFlowRun as a parameter I get the error
    AttributeError: 'StartFlowRun' object has no attribute 'result'
    When I look at the return value of StartFlowRun, it is a Task, but in the code example above it looks like StartFlowRun is supposed to return a State object. Did something change?