Description
Is your feature request related to a problem? Please describe.
As a workaround to #837 to get fully-native Python code with a happy linter, you can create Tasks
directly, but it's quite cumbersome if you also want to use @script
-decorated functions, as Task
must take a template: Optional[Union[str, Template, TemplateMixin, CallableTemplateMixin]]
, so you have to create a Script
object from the script function first, basically ignoring the decorator (and its arguments):
from hera.workflows import DAG, Script, Task, Workflow, script
@script(labels={"ignored": "args"})
def flip():
import random
result = "heads" if random.randint(0, 1) == 0 else "tails"
print(result)
with Workflow(generate_name="coinflip-", entrypoint="d") as w:
flip_template = Script(name="flip", source=flip)
with DAG(name="d") as s:
f = Task(name="flip", template=flip_template)
Describe the solution you'd like
A way to pass a script-decorated function to the Task
more succinctly for better readability while maintaining types, maybe allow template
to take a Callable
and lazily convert it to a script template? Also inferring name
automatically from the template name if not present.
from hera.workflows import DAG, Task, Workflow, script
@script(labels={"not-ignored": "args"})
def flip():
import random
result = "heads" if random.randint(0, 1) == 0 else "tails"
print(result)
with Workflow(generate_name="coinflip-", entrypoint="d") as w:
with DAG(name="d") as s:
f = Task(template=flip)
Describe alternatives you've considered
A clear and concise description of any alternative solutions or features you've considered.
Use Script
and forgo the script
decorator
Additional context
Add any other context or screenshots about the feature request here.
Trying to use template=flip
gets the following error:
examples/workflows/scripts/type_safe_coinflip.py:24: in <module>
f = Task(name="flip", template=flip)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
__pydantic_self__ = Task(), data = {'name': 'flip', 'template': <function flip at 0x1076b78b0>}, values = {'arguments': None, 'continue_on': None, 'dependencies': None, 'depends': None, ...}
fields_set = {'name', 'template'}
validation_error = ValidationError(model='Task', errors=[{'loc': ('template',), 'msg': 'str type expected', 'type': 'type_error.str'}, {'...('__root__',), 'msg': "Exactly one of ['template', 'template_ref', 'inline'] must be present", 'type': 'value_error'}])
def __init__(__pydantic_self__, **data: Any) -> None:
"""
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
"""
# Uses something other than `self` the first arg to allow "self" as a settable attribute
values, fields_set, validation_error = validate_model(__pydantic_self__.__class__, data)
if validation_error:
> raise validation_error
E pydantic.v1.error_wrappers.ValidationError: 5 validation errors for Task
E template
E str type expected (type=type_error.str)
E template
E value is not a valid dict (type=type_error.dict)
E template
E value is not a valid dict (type=type_error.dict)
E template
E value is not a valid dict (type=type_error.dict)
E __root__
E Exactly one of ['template', 'template_ref', 'inline'] must be present (type=value_error)