Skip to content

Instantly share code, notes, and snippets.

@andresgomezfrr
Created November 24, 2022 16:26
Show Gist options
  • Save andresgomezfrr/8bd105ff9c7e10e9f7854e5b13430a95 to your computer and use it in GitHub Desktop.
Save andresgomezfrr/8bd105ff9c7e10e9f7854e5b13430a95 to your computer and use it in GitHub Desktop.
from flytekit import LaunchPlan, task, workflow
from flytekit.configuration import Config
from flytekit.remote import FlyteRemote
@task(cache_version="1", cache=False) # type: ignore
def hello(name: str) -> str:
return f"Helloxz {name}"
@task(cache_version="1", cache=False) # type: ignore
def fly(name: str) -> str:
return f"Flyxz {name}"
# This is the remote task implementation
# @task(cache_version="1", cache=False) # type: ignore
# def print_empty(name: str) -> None:
# print(f"None test {str}")
remote = FlyteRemote(config=Config.auto())
@workflow # type: ignore
def workflow_test(input_name: str = "World") -> None:
hello_task = hello(name=input_name)
fly_task = fly(name=input_name)
print_remote_task_handler = remote.fetch_task(
project="andres-cookie-python-test",
domain="development",
name="andres_cookie_python_test.remote_tasks.print_empty",
)
print_task = print_remote_task_handler(name=input_name)
fly_task >> print_task
print_task >> hello_task
lp = LaunchPlan.create("remote_chain_with_remote_workflow", workflow_test)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment