acoustic-carpenter-78188
04/24/2023, 11:32 PMfrom datetime import datetime
from flytekit import task, workflow, LaunchPlan, CronSchedule
@task
def tk(k: datetime) -> str:
return str(k)
@workflow
def example_workflow(kickoff_time: datetime) -> str:
return tk(k=kickoff_time)
example_launch_plan = LaunchPlan.get_or_create(
name="example_schedule",
workflow=example_workflow,
schedule=CronSchedule(
schedule="*/5 * * * *", # Following schedule runs every min
kickoff_time_input_arg="kickoff_time", # macro magic
),
)
OR
from datetime import datetime
from flytekit import task, workflow, LaunchPlan, CronSchedule
@task
def tk2() -> str:
return "Hello"
@workflow
def example_workflow2() -> str:
return tk2()
example_launch_plan = LaunchPlan.get_or_create(
name="example_schedule2",
workflow=example_workflow2,
schedule=CronSchedule(
schedule="*/5 * * * *", # Following schedule runs every min
),
)
Where the kickoff time arguments are either present or absent.
Expected behavior
It should work in either case
Additional context to reproduce
No response
Screenshots
No response
Are you sure this issue hasn't been raised already?
☑︎ Yes
Have you read the Code of Conduct?
☑︎ Yes
flyteorg/flyteacoustic-carpenter-78188
04/24/2023, 11:32 PM