Klemens Kasseroller
12/22/2022, 10:53 AMfrom dataclasses import dataclass
from typing import Optional
from dataclasses_json import dataclass_json
from flytekit import workflow, task
from flytekit.types.file import FlyteFile
@dataclass_json
@dataclass
class TaskOutput:
file1: FlyteFile
file2: Optional[FlyteFile]
@task
def test_task_1() -> Optional[FlyteFile]:
with open("file11.txt", "w+") as f:
f.write("")
return FlyteFile("file11.txt", remote_path="<s3://test-bucket/tmp/file11.txt>")
@task
def test_task_2() -> TaskOutput:
with open("file21.txt", "w+") as f:
f.write("")
with open("file22.txt", "w+") as f:
f.write("")
return TaskOutput(
FlyteFile("file21.txt", remote_path="<s3://test-bucket/tmp/file21.txt>"),
FlyteFile("file22.txt", remote_path="<s3://test-bucket/tmp/file22.txt>")
)
@workflow
def test_workflow():
test_task_1()
test_task_2()
if __name__ == '__main__':
test_workflow()
Samhita Alla
Klemens Kasseroller
12/22/2022, 11:47 AMSamhita Alla
Eduardo Apolinario (eapolinario)
12/22/2022, 2:17 PM