https://prefect.io logo
Title
b

bolto6

08/07/2020, 12:32 PM
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,
        },
    )
n

nicholas

08/07/2020, 2:19 PM
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.