Josh
02/25/2022, 9:31 PMfrom prefect import Flow, task
from pprint import pformat
@task
def numbers():
return [1, 2]
@task
def letters():
return ["a", "b", "c"]
@task
def cross_product(numbers, letters):
res = []
for letter in letters:
for num in numbers:
res.append({"num": num, "letter": letter})
return res
@task
def pretty_print(**kwargs):
print(pformat(kwargs))
with Flow("...") as flow:
n = numbers()
l = letters()
map_items = cross_product(n, l)
pretty_print.map(map_items)
flow.run()
I get the following error:
Traceback (most recent call last):
File "flow.py", line 33, in <module>
pretty_print.map(map_items)
File "prefect/core/task.py", line 760, in map
*args, mapped=True, upstream_tasks=upstream_tasks, flow=flow, **kwargs
File "prefect/core/task.py", line 674, in bind
callargs = dict(signature.bind(*args, **kwargs).arguments) # type: Dict
File "inspect.py", line 3015, in bind
return args[0]._bind(args[1:], kwargs)
File "inspect.py", line 2942, in _bind
'too many positional arguments') from None
TypeError: too many positional arguments
Kevin Kho
02/25/2022, 9:34 PMJosh
02/25/2022, 9:49 PMitems
dictionary. Can we dynamically rename the task with task_run_name="number: {items[num]} letter: {items[letter]}"
?Kevin Kho
02/25/2022, 9:50 PM