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
04/19/2022, 9:20 PMTim Wright
04/19/2022, 9:20 PMKevin Kho
04/19/2022, 9:21 PMTim Wright
04/19/2022, 9:26 PMKevin Kho
04/19/2022, 9:31 PM@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
04/19/2022, 9:33 PM@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
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
04/19/2022, 9:35 PM