b

    bolto6

    2 years ago
    How i can add parameters in
    b_flow
    from
    a_flow
    and dir parameter? Flow A:
    python
    @task
    def any_work(a: str) -> str:
        return f'/tmp/{a}'
    
    with Flow('a_flow') as flow:
        name = Parameter('name', default='all')
        result = any_work(name)
    Flow B:
    python
    @task
    def any_work(a: str, b: str) -> str:
        return f'{a}/{b}'
    
    with Flow('b_flow') as flow:
        home = Parameter('home', required=True)
        dir = Parameter('dir', default='any')
    
        result = any_work(home, dir)
    Flow C:
    python
    a_flow = FlowRunTask(flow_name='a_flow', wait=True)
    b_flow = FlowRunTask(flow_name='b_flow', wait=True)
    
    with Flow('c_flow') as flow:
        name = Parameter('name', default='all')
        dir = Parameter('dir', default='any')
    
        a_flow_state = a_flow(parameters={'name': name})
    
        # this error code, but how i can add parameters in `b_flow` from `a_flow` and dir parameter?
        result = b_flow(
            upstream_tasks=[a_flow_state],
            parameters={
                'home': a_flow_state.result,
                'dir': dir,
            },
        )
    nicholas

    nicholas

    2 years ago
    Hi @bolto6 - my recommendation would be to use an independent results location to store/retrieve data between your flows or to use the Flow Run Task to run the next Flow in your pipeline with the parameters of your choosing. Note that you'll need to call FlowRunTask from within the Flow context, so something like this:
    @task
    def any_work(a: str) -> str:
        return f'/tmp/{a}'
    
    @task
    def any_work(a: str, b: str) -> str:
        return f'{a}/{b}'
    
    with Flow('a_flow') as flow:
        name = Parameter('name', default='all')
        result = any_work(name)
    
        FlowRunTask(flow_name="b_flow", parameters={'home': result})
    etc.