feat(bindings): Task arguments default value binding

Resolves: flyteorg/flyte#5321
Signed-off-by: Chi-Sheng Liu <[email protected]>

flytekit/core/promise.py

@@ -1064,7 +1064,11 @@ def create_and_link_node(
  f"The default value for the optional type must be None, but got {_default}"
  )
  is_optional = True
- if not is_optional:
+ if is_optional:
+ continue
+ if k in interface.inputs_with_defaults:
+ kwargs[k] = interface.inputs_with_defaults[k][1]
+ else:
  from flytekit.core.base_task import Task
 
  error_msg = f"Input {k} of type {interface.inputs[k]} was not specified for function {entity.name}"
@@ -1077,8 +1081,6 @@ def create_and_link_node(
  )
 
  raise _user_exceptions.FlyteAssertion(error_msg)
- else:
- continue
  v = kwargs[k]
  # This check ensures that tuples are not passed into a function, as tuples are not supported by Flyte
  # Usually a Tuple will indicate that multiple outputs from a previous task were accidentally passed

tests/flytekit/unit/core/test_promise.py

@@ -46,6 +46,14 @@ def t2(a: typing.Optional[int] = None) -> typing.Optional[int]:
  assert p.ref.var == "o0"
  assert len(p.ref.node.bindings) == 0
 
+ @task
+ def t_default_value(a: int = 1) -> typing.Optional[int]:
+ return a
+
+ p = create_and_link_node(ctx, t_default_value)
+ assert p.ref.var == "o0"
+ assert len(p.ref.node.bindings) == 1
+
 
 def test_create_and_link_node_from_remote():
  @task