https://prefect.io logo
Title
o

Oliver Mannion

07/01/2022, 8:02 AM
Hiya, is it possible to use a string concatenation of a task result as an input parameter to a downstream task? 🧵
@task
def determine_batch_start_and_end() -> Tuple[str, str]:
    now = pendulum.now(tz="UTC")
    batch_start, batch_end = get_last_hour(now)
    return (batch_start, batch_end)

def build_flow(flow_name: str) -> Flow:
    with Flow(flow_name) as flow:

        batch_start, batch_end = determine_batch_start_and_end()

        create_flow_run(
            flow_id=settings.flow_id,
            run_name=f"br_{str(batch_start)}"
        )

    return flow
I'd like
run_name
to be something like
br_2022-07-01 07:00:00
but it ends up being
br_<Task: determine_batch_start_and_end[0]>
a

Anna Geller

07/01/2022, 11:26 AM
you would need to solve it using separate task as described here https://discourse.prefect.io/t/how-to-dynamically-change-the-run-name-in-the-create-flow-run-task/97#solution-1
:thank-you: 1