Use TaskSpec instead of TaskTemplate for fetch_task and avoid network when loading module

flytekit/clients/raw.py

@@ -79,7 +79,6 @@ def handler(self, create_request):
                 cli_logger.error(_MessageToJson(create_request))
                 cli_logger.error("Details returned from the flyte admin: ")
                 cli_logger.error(e.details)
-                e.details += "create_request: " + _MessageToJson(create_request)
             # Re-raise since we're not  handling the error here and add the create_request details
             raise e
 
@@ -260,7 +259,6 @@ def _refresh_credentials_from_command(self):
         :param self: RawSynchronousFlyteClient
         :return:
         """
-
         command = self._cfg.command
         if not command:
             raise FlyteAuthenticationException("No command specified in configuration for command authentication")

flytekit/clis/sdk_in_container/register.py

@@ -107,6 +107,12 @@
     is_flag=True,
     help="Enables to skip zipping and uploading the package",
 )
+@click.option(
+    "--dry-run",
+    default=False,
+    is_flag=True,
+    help="Execute registration in dry-run mode. Skips actual registration to remote",
+)
 @click.argument("package-or-module", type=click.Path(exists=True, readable=True, resolve_path=True), nargs=-1)
 @click.pass_context
 def register(
@@ -122,6 +128,7 @@ def register(
     deref_symlinks: bool,
     non_fast: bool,
     package_or_module: typing.Tuple[str],
+    dry_run: bool,
 ):
     """
     see help
@@ -156,6 +163,7 @@ def register(
 
     # Create and save FlyteRemote,
     remote = get_and_save_remote_with_click_context(ctx, project, domain)
+    click.secho(f"Registering against {remote.config.platform.endpoint}")
     try:
         repo.register(
             project,
@@ -170,6 +178,7 @@ def register(
             fast=not non_fast,
             package_or_module=package_or_module,
             remote=remote,
+            dry_run=dry_run,
         )
     except Exception as e:
         raise e

flytekit/core/node.py

@@ -51,6 +51,10 @@ def __rshift__(self, other: Node):
         self.runs_before(other)
         return other
 
+    @property
+    def name(self) -> str:
+        return self._id
+
     @property
     def outputs(self):
         if self._outputs is None:

flytekit/core/promise.py

@@ -870,7 +870,6 @@ def create_and_link_node_from_remote(
     )
 
     flytekit_node = Node(
-        # TODO: Better naming, probably a derivative of the function name.
         id=f"{ctx.compilation_state.prefix}n{len(ctx.compilation_state.nodes)}",
         metadata=entity.construct_node_metadata(),
         bindings=sorted(bindings, key=lambda b: b.var),

flytekit/core/python_function_task.py

@@ -217,12 +217,7 @@ def compile_into_workflow(
             for entity, model in model_entities.items():
                 # We only care about gathering tasks here. Launch plans are handled by
                 # propeller. Subworkflows should already be in the workflow spec.
-                if not isinstance(entity, Task) and not isinstance(entity, task_models.TaskTemplate):
-                    continue
-
-                # Handle FlyteTask
-                if isinstance(entity, task_models.TaskTemplate):
-                    tts.append(entity)
+                if not isinstance(entity, Task) and not isinstance(entity, task_models.TaskSpec):
                     continue
 
                 # We are currently not supporting reference tasks since these will

flytekit/remote/__init__.py

@@ -85,10 +85,14 @@
 
 """
 
-from flytekit.remote.component_nodes import FlyteTaskNode, FlyteWorkflowNode
+from flytekit.remote.entities import (
+    FlyteBranchNode,
+    FlyteLaunchPlan,
+    FlyteNode,
+    FlyteTask,
+    FlyteTaskNode,
+    FlyteWorkflow,
+    FlyteWorkflowNode,
+)
 from flytekit.remote.executions import FlyteNodeExecution, FlyteTaskExecution, FlyteWorkflowExecution
-from flytekit.remote.launch_plan import FlyteLaunchPlan
-from flytekit.remote.nodes import FlyteNode
 from flytekit.remote.remote import FlyteRemote
-from flytekit.remote.task import FlyteTask
-from flytekit.remote.workflow import FlyteWorkflow