update typing hints for Tuple (#465)

* update task cache docs (#459)

Signed-off-by: Niels Bantilan <niels.bantilan@gmail.com>
Signed-off-by: Qiwen Yu <yvaineyuu.0201@gmail.com>

* update typing for Tuple

fix issue #1710
flyteorg/flyte#1710
change typing hints from ( , ) to Tuple[ , ]

Signed-off-by: Qiwen Yu <yvaineyuu.0201@gmail.com>

* update more Tuple typing hints

Signed-off-by: Qiwen Yu <yvaineyuu.0201@gmail.com>

Co-authored-by: Niels Bantilan <niels.bantilan@gmail.com>

cookbook/case_studies/ml_training/house_price_prediction/house_price_predictor.py

@@ -25,6 +25,7 @@
 from xgboost import XGBRegressor
 from flytekit import Resources, dynamic, task, workflow
 from flytekit.types.file import FlyteFile
+from typing import Tuple
 
 # %%
 # Initializing the Variables
@@ -96,7 +97,7 @@ def gen_houses(num_houses) -> pd.DataFrame:
 # Split the data into train, val, and test datasets.
 def split_data(
     df: pd.DataFrame, seed: int, split: typing.List[float]
-) -> (pd.DataFrame, pd.DataFrame, pd.DataFrame):
+) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
 
     seed = seed
     val_size = split[1]

cookbook/case_studies/ml_training/pima_diabetes/diabetes.py

@@ -20,6 +20,7 @@
 from sklearn.metrics import accuracy_score
 from sklearn.model_selection import train_test_split
 from xgboost import XGBClassifier
+from typing import Tuple
 
 # %%
 # Since we are working with a specific dataset, we will create a strictly typed schema for the dataset.
@@ -70,12 +71,12 @@
 @task(cache_version="1.0", cache=True, limits=Resources(mem="200Mi"))
 def split_traintest_dataset(
     dataset: FlyteFile[typing.TypeVar("csv")], seed: int, test_split_ratio: float
-) -> (
+) -> Tuple[
     FlyteSchema[FEATURE_COLUMNS],
     FlyteSchema[FEATURE_COLUMNS],
     FlyteSchema[CLASSES_COLUMNS],
     FlyteSchema[CLASSES_COLUMNS],
-):
+]:
     """
     Retrieves the training dataset from the given blob location and then splits it using the split ratio and returns the result
     This splitter is only for the dataset that has the format as specified in the example csv. The last column is assumed to be

cookbook/core/containerization/use_secrets.py

@@ -67,6 +67,7 @@
 # %%
 import os
 import flytekit
+from typing import Tuple
 
 # %%
 # Flytekit exposes a type/class called Secrets. It can be imported as follows.
@@ -120,7 +121,7 @@ def secret_task() -> str:
 # The Secret structure allows passing two fields, matching the key and the group, as previously described:
 @task(
     secret_requests=[Secret(key=USERNAME_SECRET, group=SECRET_GROUP), Secret(key=PASSWORD_SECRET, group=SECRET_GROUP)])
-def user_info_task() -> (str, str):
+def user_info_task() -> Tuple[str, str]:
     secret_username = flytekit.current_context().secrets.get(SECRET_GROUP, USERNAME_SECRET)
     secret_pwd = flytekit.current_context().secrets.get(SECRET_GROUP, PASSWORD_SECRET)
     # Please do not print the secret value, this is just a demonstration.
@@ -135,7 +136,7 @@ def user_info_task() -> (str, str):
 # In these scenarios you can specify the mount_requirement. In the following example we force the mounting to be
 # an Env variable
 @task(secret_requests=[Secret(group=SECRET_GROUP, key=SECRET_NAME, mount_requirement=Secret.MountType.ENV_VAR)])
-def secret_file_task() -> (str, str):
+def secret_file_task() -> Tuple[str, str]:
     # SM here is a handle to the secrets manager
     sm = flytekit.current_context().secrets
     f = sm.get_secrets_file(SECRET_GROUP, SECRET_NAME)
@@ -147,7 +148,7 @@ def secret_file_task() -> (str, str):
 # %%
 # These tasks can be used in your workflow as usual
 @workflow
-def my_secret_workflow() -> (str, str, str, str, str):
+def my_secret_workflow() -> Tuple[str, str, str, str, str]:
     x = secret_task()
     y, z = user_info_task()
     f, s = secret_file_task()

cookbook/core/control_flow/run_merge_sort.py

@@ -17,6 +17,7 @@
 from random import random, seed
 
 from flytekit import conditional, dynamic, task, workflow
+from typing import Tuple
 
 # seed random number generator
 seed(datetime.now().microsecond)
@@ -25,7 +26,7 @@
 # %%
 # A simple split function that divides a list into two halves.
 @task
-def split(numbers: typing.List[int]) -> (typing.List[int], typing.List[int], int):
+def split(numbers: typing.List[int]) -> Tuple[typing.List[int], typing.List[int], int]:
     return (
         numbers[0:int(len(numbers) / 2)],
         numbers[int(len(numbers) / 2):],

cookbook/core/control_flow/subworkflows.py

@@ -35,7 +35,7 @@
 """
 
 import typing
-
+from typing import Tuple
 from flytekit import task, workflow
 
 
@@ -54,7 +54,7 @@ def t1(a: int) -> op:
 # This will be the subworkflow of our examples, but note that this is a workflow like any other. It can be run just
 # like any other workflow. Note here that the workflow has been declared with a default.
 @workflow
-def my_subwf(a: int = 42) -> (str, str):
+def my_subwf(a: int = 42) -> Tuple[str, str]:
     x, y = t1(a=a)
     u, v = t1(a=x)
     return y, v
@@ -74,7 +74,7 @@ def my_subwf(a: int = 42) -> (str, str):
 #
 #    Also note the use of with_overrides to provide a new name to the graph-node for better rendering or readability
 @workflow
-def parent_wf(a: int) -> (int, str, str):
+def parent_wf(a: int) -> Tuple[int, str, str]:
     x, y = t1(a=a).with_overrides(node_name="node-t1-parent")
     u, v = my_subwf(a=x)
     return x, u, v
@@ -93,7 +93,7 @@ def parent_wf(a: int) -> (int, str, str):
 # can be simply composed from other workflows, even if the other workflows are standalone entities. Each of the
 # workflows in this module can exist independently and executed independently
 @workflow
-def nested_parent_wf(a: int) -> (int, str, str, str):
+def nested_parent_wf(a: int) -> Tuple[int, str, str, str]:
     x, y = my_subwf(a=a)
     m, n, o = parent_wf(a=a)
     return m, n, o, y

cookbook/core/flyte_basics/basic_workflow.py

@@ -19,6 +19,7 @@
 import typing
 
 from flytekit import task, workflow
+from typing import Tuple
 
 
 @task
@@ -35,7 +36,7 @@ def t2(a: str, b: str) -> str:
 # You can treat the outputs of a task as you normally would a Python function. Assign the output to two variables
 # and use them in subsequent tasks as normal. See :py:func:`flytekit.workflow`
 @workflow
-def my_wf(a: int, b: str) -> (int, str):
+def my_wf(a: int, b: str) -> Tuple[int, str]:
     x, y = t1(a=a)
     d = t2(a=y, b=b)
     return x, d

cookbook/core/type_system/enums.py

@@ -17,7 +17,7 @@
 valid default.
 """
 from flytekit import task, workflow
-import typing
+from typing import Tuple
 from enum import Enum
 
 
@@ -50,7 +50,7 @@ def string_to_enum(c: str) -> Color:
 
 
 @workflow
-def enum_wf(c: Color = Color.RED) -> (Color, str):
+def enum_wf(c: Color = Color.RED) -> Tuple[Color, str]:
     v = enum_stringify(c=c)
     return string_to_enum(c=v), v
 

cookbook/integrations/aws/sagemaker_training/sagemaker_custom_training.py

@@ -11,6 +11,7 @@
 import tensorflow_datasets as tfds
 from flytekit import task, workflow
 from flytekit.types.directory import TensorboardLogs
+from typing import Tuple
 
 # %%
 # Training Algorithm
@@ -184,7 +185,7 @@ def plot_loss_and_accuracy(epoch_logs: dict) -> PlotOutputs:
 @workflow
 def mnist_trainer(
     epochs: int = 5, batch_size: int = 128
-) -> (HDF5EncodedFile, PNGImageFile, PNGImageFile, TensorboardLogs):
+) -> Tuple[HDF5EncodedFile, PNGImageFile, PNGImageFile, TensorboardLogs]:
     model, history, logs = custom_training_task(epochs=epochs, batch_size=batch_size)
     accuracy, loss = plot_loss_and_accuracy(epoch_logs=history)
     return model, accuracy, loss, logs

cookbook/integrations/kubernetes/kfpytorch/pytorch_mnist.py

@@ -22,6 +22,7 @@
 from torch import distributed as dist
 from torch import nn, optim
 from torchvision import datasets, transforms
+from typing import Tuple
 
 WORLD_SIZE = int(os.environ.get("WORLD_SIZE", 1))
 
@@ -297,7 +298,7 @@ def plot_accuracy(epoch_accuracies: typing.List[float]) -> PNGImageFile:
 @workflow
 def pytorch_training_wf(
     hp: Hyperparameters,
-) -> (PythonPickledFile, PNGImageFile, TensorboardLogs):
+) -> Tuple[PythonPickledFile, PNGImageFile, TensorboardLogs]:
     accuracies, model, logs = mnist_pytorch_job(hp=hp)
     plot = plot_accuracy(epoch_accuracies=accuracies)
     return model, plot, logs