Tim Wright
04/19/2022, 9:16 PMdict_result = some_task()
some_other_task(**dict_result)
Is this doable? Is there a more Prefect-y way to accomplish this without creating a Parameter for eahc element of the dictionary? I ultimately want to pass as an input a list of dictionaries which I can then operate over using Prefect's mapping, so being able to define the collection of arguments together is important.Kevin Kho
Tim Wright
04/19/2022, 9:20 PMKevin Kho
Kevin Kho
Tim Wright
04/19/2022, 9:26 PMKevin Kho
@task
def get_item(raw: List[Dict], key):
res = []
for r in raw:
res.append(r[key])
return res
with Flow(...) as flow:
items = get_all_inputs()
field1 = get_item(items, "field1")
field2 = get_item(items, "field2")
next_task.map(field1, field2)
Tim Wright
04/19/2022, 9:32 PMKevin Kho
@task(nout=2)
def get_item(raw: List[Dict], key1, key2):
res1 = []
res2 = []
for r in raw:
res1.append(r[key1])
res2.append(r[key2])
return res1, res2
Kevin Kho
with Flow(...) as flow:
items = get_all_inputs()
field1, field2 = get_item(items, "field1", "field2")
next_task.map(field1, field2)
Tim Wright
04/19/2022, 9:35 PMKevin Kho