Merge branch 'master' into fix-remote-docs

CODEOWNERS

@@ -1,3 +1,3 @@
 # These owners will be the default owners for everything in
 # the repo. Unless a later match takes precedence.
-*       @wild-endeavor @kumare3 @eapolinario @pingsutw
+*       @wild-endeavor @kumare3 @eapolinario @pingsutw @cosmicBboy

dev-requirements.txt

@@ -531,7 +531,7 @@ tomli==2.0.1
     #   coverage
     #   mypy
     #   pytest
-torch==1.12.1
+torch==1.13.1
     # via -r dev-requirements.in
 traitlets==5.6.0
     # via

flytekit/__init__.py

@@ -190,6 +190,7 @@
 from flytekit.models.common import Annotations, AuthRole, Labels
 from flytekit.models.core.execution import WorkflowExecutionPhase
 from flytekit.models.core.types import BlobType
+from flytekit.models.documentation import Description, Documentation, SourceCode
 from flytekit.models.literals import Blob, BlobMetadata, Literal, Scalar
 from flytekit.models.types import LiteralType
 from flytekit.types import directory, file, numpy, schema

flytekit/clients/raw.py

@@ -10,11 +10,13 @@
 import grpc
 import requests as _requests
 from flyteidl.admin.project_pb2 import ProjectListRequest
+from flyteidl.admin.signal_pb2 import SignalList, SignalListRequest, SignalSetRequest, SignalSetResponse
 from flyteidl.service import admin_pb2_grpc as _admin_service
 from flyteidl.service import auth_pb2
 from flyteidl.service import auth_pb2_grpc as auth_service
 from flyteidl.service import dataproxy_pb2 as _dataproxy_pb2
 from flyteidl.service import dataproxy_pb2_grpc as dataproxy_service
+from flyteidl.service import signal_pb2_grpc as signal_service
 from flyteidl.service.dataproxy_pb2_grpc import DataProxyServiceStub
 from google.protobuf.json_format import MessageToJson as _MessageToJson
 
@@ -145,6 +147,7 @@ def __init__(self, cfg: PlatformConfig, **kwargs):
             )
         self._stub = _admin_service.AdminServiceStub(self._channel)
         self._auth_stub = auth_service.AuthMetadataServiceStub(self._channel)
+        self._signal = signal_service.SignalServiceStub(self._channel)
         try:
             resp = self._auth_stub.GetPublicClientConfig(auth_pb2.PublicClientAuthConfigRequest())
             self._public_client_config = resp
@@ -406,6 +409,20 @@ def get_task(self, get_object_request):
         """
         return self._stub.GetTask(get_object_request, metadata=self._metadata)
 
+    @_handle_rpc_error(retry=True)
+    def set_signal(self, signal_set_request: SignalSetRequest) -> SignalSetResponse:
+        """
+        This sets a signal
+        """
+        return self._signal.SetSignal(signal_set_request, metadata=self._metadata)
+
+    @_handle_rpc_error(retry=True)
+    def list_signals(self, signal_list_request: SignalListRequest) -> SignalList:
+        """
+        This lists signals
+        """
+        return self._signal.ListSignals(signal_list_request, metadata=self._metadata)
+
     ####################################################################################################################
     #
     #  Workflow Endpoints

flytekit/clis/sdk_in_container/register.py

@@ -17,7 +17,7 @@
 and the flytectl register step in one command. This is why you see switches you'd normally use with flytectl
 like service account here.
 
-Note: This command runs "fast" register by default. Future work to come to add a non-fast version.
+Note: This command runs "fast" register by default.
 This means that a zip is created from the detected root of the packages given, and uploaded. Just like with
 pyflyte run, tasks registered from this command will download and unzip that code package before running.
 

flytekit/clis/sdk_in_container/run.py

@@ -303,6 +303,9 @@ def convert_to_literal(
         if self._literal_type.simple or self._literal_type.enum_type:
             if self._literal_type.simple and self._literal_type.simple == SimpleType.STRUCT:
                 if self._python_type == dict:
+                    if type(value) != str:
+                        # The type of default value is dict, so we have to convert it to json string
+                        value = json.dumps(value)
                     o = json.loads(value)
                 elif type(value) != self._python_type:
                     o = cast(DataClassJsonMixin, self._python_type).from_json(value)

flytekit/configuration/__init__.py

@@ -53,6 +53,7 @@
    ~Image
    ~ImageConfig
    ~SerializationSettings
+   ~FastSerializationSettings
 
 .. _configuration-execution-time-settings:
 
@@ -462,7 +463,7 @@ class GCSConfig(object):
     gsutil_parallelism: bool = False
 
     @classmethod
-    def auto(self, config_file: typing.Union[str, ConfigFile] = None) -> GCSConfig:
+    def auto(cls, config_file: typing.Union[str, ConfigFile] = None) -> GCSConfig:
         config_file = get_config_file(config_file)
         kwargs = {}
         kwargs = set_if_exists(kwargs, "gsutil_parallelism", _internal.GCP.GSUTIL_PARALLELISM.read(config_file))
@@ -646,6 +647,7 @@ class SerializationSettings(object):
     domain: typing.Optional[str] = None
     version: typing.Optional[str] = None
     env: Optional[Dict[str, str]] = None
+    git_repo: Optional[str] = None
     python_interpreter: str = DEFAULT_RUNTIME_PYTHON_INTERPRETER
     flytekit_virtualenv_root: Optional[str] = None
     fast_serialization_settings: Optional[FastSerializationSettings] = None
@@ -718,6 +720,7 @@ def new_builder(self) -> Builder:
             version=self.version,
             image_config=self.image_config,
             env=self.env.copy() if self.env else None,
+            git_repo=self.git_repo,
             flytekit_virtualenv_root=self.flytekit_virtualenv_root,
             python_interpreter=self.python_interpreter,
             fast_serialization_settings=self.fast_serialization_settings,
@@ -767,6 +770,7 @@ class Builder(object):
         version: str
         image_config: ImageConfig
         env: Optional[Dict[str, str]] = None
+        git_repo: Optional[str] = None
         flytekit_virtualenv_root: Optional[str] = None
         python_interpreter: Optional[str] = None
         fast_serialization_settings: Optional[FastSerializationSettings] = None
@@ -782,6 +786,7 @@ def build(self) -> SerializationSettings:
                 version=self.version,
                 image_config=self.image_config,
                 env=self.env,
+                git_repo=self.git_repo,
                 flytekit_virtualenv_root=self.flytekit_virtualenv_root,
                 python_interpreter=self.python_interpreter,
                 fast_serialization_settings=self.fast_serialization_settings,

flytekit/configuration/file.py

@@ -224,7 +224,7 @@ def legacy_config(self) -> _configparser.ConfigParser:
         return self._legacy_config
 
     @property
-    def yaml_config(self) -> typing.Dict[str, Any]:
+    def yaml_config(self) -> typing.Dict[str, typing.Any]:
         return self._yaml_config
 
 

flytekit/core/base_task.py

@@ -45,6 +45,7 @@
 from flytekit.models import literals as _literal_models
 from flytekit.models import task as _task_model
 from flytekit.models.core import workflow as _workflow_model
+from flytekit.models.documentation import Description, Documentation
 from flytekit.models.interface import Variable
 from flytekit.models.security import SecurityContext
 
@@ -156,6 +157,7 @@ def __init__(
         metadata: Optional[TaskMetadata] = None,
         task_type_version=0,
         security_ctx: Optional[SecurityContext] = None,
+        docs: Optional[Documentation] = None,
         **kwargs,
     ):
         self._task_type = task_type
@@ -164,6 +166,7 @@ def __init__(
         self._metadata = metadata if metadata else TaskMetadata()
         self._task_type_version = task_type_version
         self._security_ctx = security_ctx
+        self._docs = docs
 
         FlyteEntities.entities.append(self)
 
@@ -195,6 +198,10 @@ def task_type_version(self) -> int:
     def security_context(self) -> SecurityContext:
         return self._security_ctx
 
+    @property
+    def docs(self) -> Documentation:
+        return self._docs
+
     def get_type_for_input_var(self, k: str, v: Any) -> type:
         """
         Returns the python native type for the given input variable
@@ -390,6 +397,17 @@ def __init__(
         self._environment = environment if environment else {}
         self._task_config = task_config
         self._disable_deck = disable_deck
+        if self._python_interface.docstring:
+            if self.docs is None:
+                self._docs = Documentation(
+                    short_description=self._python_interface.docstring.short_description,
+                    long_description=Description(value=self._python_interface.docstring.long_description),
+                )
+            else:
+                if self._python_interface.docstring.short_description:
+                    self._docs.short_description = self._python_interface.docstring.short_description
+                if self._python_interface.docstring.long_description:
+                    self._docs.long_description = Description(value=self._python_interface.docstring.long_description)
 
     # TODO lets call this interface and the other as flyte_interface?
     @property

flytekit/core/interface.py

@@ -207,7 +207,6 @@ def transform_interface_to_typed_interface(
     """
     if interface is None:
         return None
-
     if interface.docstring is None:
         input_descriptions = output_descriptions = {}
     else:

flytekit/core/task.py

@@ -7,6 +7,7 @@
 from flytekit.core.python_function_task import PythonFunctionTask
 from flytekit.core.reference_entity import ReferenceEntity, TaskReference
 from flytekit.core.resources import Resources
+from flytekit.models.documentation import Documentation
 from flytekit.models.security import Secret
 
 
@@ -89,6 +90,7 @@ def task(
     secret_requests: Optional[List[Secret]] = None,
     execution_mode: Optional[PythonFunctionTask.ExecutionBehavior] = PythonFunctionTask.ExecutionBehavior.DEFAULT,
     task_resolver: Optional[TaskResolverMixin] = None,
+    docs: Optional[Documentation] = None,
     disable_deck: bool = True,
 ) -> Union[Callable, PythonFunctionTask]:
     """
@@ -179,6 +181,7 @@ def foo2():
     :param execution_mode: This is mainly for internal use. Please ignore. It is filled in automatically.
     :param task_resolver: Provide a custom task resolver.
     :param disable_deck: If true, this task will not output deck html file
+    :param docs: Documentation about this task
     """
 
     def wrapper(fn) -> PythonFunctionTask:
@@ -204,6 +207,7 @@ def wrapper(fn) -> PythonFunctionTask:
             execution_mode=execution_mode,
             task_resolver=task_resolver,
             disable_deck=disable_deck,
+            docs=docs,
         )
         update_wrapper(task_instance, fn)
         return task_instance

flytekit/core/type_engine.py

@@ -361,6 +361,12 @@ def _serialize_flyte_type(self, python_val: T, python_type: Type[T]) -> typing.A
         from flytekit.types.schema.types import FlyteSchema
         from flytekit.types.structured.structured_dataset import StructuredDataset
 
+        # Handle Optional
+        if get_origin(python_type) is typing.Union and type(None) in get_args(python_type):
+            if python_val is None:
+                return None
+            return self._serialize_flyte_type(python_val, get_args(python_type)[0])
+
         if hasattr(python_type, "__origin__") and python_type.__origin__ is list:
             return [self._serialize_flyte_type(v, python_type.__args__[0]) for v in python_val]
 
@@ -400,12 +406,18 @@ def _serialize_flyte_type(self, python_val: T, python_type: Type[T]) -> typing.A
                 python_val.__setattr__(v.name, self._serialize_flyte_type(val, field_type))
             return python_val
 
-    def _deserialize_flyte_type(self, python_val: T, expected_python_type: Type) -> T:
+    def _deserialize_flyte_type(self, python_val: T, expected_python_type: Type) -> Optional[T]:
         from flytekit.types.directory.types import FlyteDirectory, FlyteDirToMultipartBlobTransformer
         from flytekit.types.file.file import FlyteFile, FlyteFilePathTransformer
         from flytekit.types.schema.types import FlyteSchema, FlyteSchemaTransformer
         from flytekit.types.structured.structured_dataset import StructuredDataset, StructuredDatasetTransformerEngine
 
+        # Handle Optional
+        if get_origin(expected_python_type) is typing.Union and type(None) in get_args(expected_python_type):
+            if python_val is None:
+                return None
+            return self._deserialize_flyte_type(python_val, get_args(expected_python_type)[0])
+
         if hasattr(expected_python_type, "__origin__") and expected_python_type.__origin__ is list:
             return [self._deserialize_flyte_type(v, expected_python_type.__args__[0]) for v in python_val]  # type: ignore
 

flytekit/core/workflow.py

@@ -39,6 +39,7 @@
 from flytekit.models import interface as _interface_models
 from flytekit.models import literals as _literal_models
 from flytekit.models.core import workflow as _workflow_model
+from flytekit.models.documentation import Description, Documentation
 
 GLOBAL_START_NODE = Node(
     id=_common_constants.GLOBAL_INPUT_NODE_ID,
@@ -168,6 +169,7 @@ def __init__(
         workflow_metadata: WorkflowMetadata,
         workflow_metadata_defaults: WorkflowMetadataDefaults,
         python_interface: Interface,
+        docs: Optional[Documentation] = None,
         **kwargs,
     ):
         self._name = name
@@ -179,13 +181,31 @@ def __init__(
         self._unbound_inputs = set()
         self._nodes = []
         self._output_bindings: List[_literal_models.Binding] = []
+        self._docs = docs
+
+        if self._python_interface.docstring:
+            if self.docs is None:
+                self._docs = Documentation(
+                    short_description=self._python_interface.docstring.short_description,
+                    long_description=Description(value=self._python_interface.docstring.long_description),
+                )
+            else:
+                if self._python_interface.docstring.short_description:
+                    self._docs.short_description = self._python_interface.docstring.short_description
+                if self._python_interface.docstring.long_description:
+                    self._docs = Description(value=self._python_interface.docstring.long_description)
+
         FlyteEntities.entities.append(self)
         super().__init__(**kwargs)
 
     @property
     def name(self) -> str:
         return self._name
 
+    @property
+    def docs(self):
+        return self._docs
+
     @property
     def short_name(self) -> str:
         return extract_obj_name(self._name)
@@ -571,7 +591,8 @@ def __init__(
         workflow_function: Callable,
         metadata: Optional[WorkflowMetadata],
         default_metadata: Optional[WorkflowMetadataDefaults],
-        docstring: Docstring = None,
+        docstring: Optional[Docstring] = None,
+        docs: Optional[Documentation] = None,
     ):
         name, _, _, _ = extract_task_module(workflow_function)
         self._workflow_function = workflow_function
@@ -586,6 +607,7 @@ def __init__(
             workflow_metadata=metadata,
             workflow_metadata_defaults=default_metadata,
             python_interface=native_interface,
+            docs=docs,
         )
 
     @property
@@ -690,6 +712,7 @@ def workflow(
     _workflow_function=None,
     failure_policy: Optional[WorkflowFailurePolicy] = None,
     interruptible: bool = False,
+    docs: Optional[Documentation] = None,
 ):
     """
     This decorator declares a function to be a Flyte workflow. Workflows are declarative entities that construct a DAG
@@ -718,6 +741,7 @@ def workflow(
     :param _workflow_function: This argument is implicitly passed and represents the decorated function.
     :param failure_policy: Use the options in flytekit.WorkflowFailurePolicy
     :param interruptible: Whether or not tasks launched from this workflow are by default interruptible
+    :param docs: Description entity for the workflow
     """
 
     def wrapper(fn):
@@ -730,6 +754,7 @@ def wrapper(fn):
             metadata=workflow_metadata,
             default_metadata=workflow_metadata_defaults,
             docstring=Docstring(callable_=fn),
+            docs=docs,
         )
         workflow_instance.compile()
         update_wrapper(workflow_instance, fn)