Frank Shen
10/10/2022, 7:00 PMfrom flytekit import LaunchPlan, current_context, task, workflow
@task
def square(val: int) -> int:
return val * val
@workflow
def my_wf(val: int) -> int:
result = square(val=val)
return result
default_lp = LaunchPlan.get_default_launch_plan(current_context(), my_wf)
square_3 = default_lp(val=3)
Kevin Su
10/10/2022, 9:14 PMLaunchPlan.get_or_create
.check this docFrank Shen
10/10/2022, 9:53 PMKevin Su
10/10/2022, 9:56 PMpyflyte register example.py
. it will register your LP.Frank Shen
10/10/2022, 10:28 PMkickoff_time_input_arg="kickoff_time"
in
schedule=CronSchedule(
schedule="*/1 * * * *",
kickoff_time_input_arg="kickoff_time",
)
Kevin Su
10/10/2022, 10:33 PMFrank Shen
10/10/2022, 10:49 PMKevin Su
10/10/2022, 11:07 PMflytectl sandbox start
insteadFrank Shen
10/10/2022, 11:09 PMSamhita Alla
Frank Shen
10/11/2022, 5:00 PM@workflow
def my_wf(
# kickoff_time: datetime,
val: int
) -> int:
# print(f"datetime run: {kickoff_time}")
result = square(val=val)
return result
schedule = CronSchedule(
schedule="*/1 * * * *",
# kickoff_time_input_arg="kickoff_time"
)
lp = LaunchPlan.get_or_create(
workflow=my_wf, name="lp_test_1", schedule = schedule, default_inputs={"val": 3}
)
To register I used
pyflyte register lp-test-1.py
Samhita Alla
flytectl update launchplan -p flytesnacks -d development <lp-name> --version <version> --activate
Frank Shen
10/12/2022, 4:20 PM