Remove legacy API

flytekit/__init__.py

@@ -160,7 +160,6 @@
 else:
     from importlib.metadata import entry_points
 
-import flytekit.plugins  # This will be deprecated, these are the old plugins, the new plugins live in plugins/
 from flytekit.core.base_sql_task import SQLTask
 from flytekit.core.base_task import SecurityContext, TaskMetadata, kwtypes
 from flytekit.core.condition import conditional

flytekit/bin/entrypoint.py

@@ -1,29 +1,23 @@
 import contextlib
 import datetime as _datetime
-import importlib as _importlib
 import logging as python_logging
 import os as _os
 import pathlib
-import random as _random
 import traceback as _traceback
 from typing import List
 
 import click as _click
 from flyteidl.core import literals_pb2 as _literals_pb2
 
 from flytekit import PythonFunctionTask
-from flytekit.common import constants as _constants
-from flytekit.common import utils as _common_utils
-from flytekit.common import utils as _utils
-from flytekit.common.exceptions import scopes as _scoped_exceptions
-from flytekit.common.exceptions import scopes as _scopes
-from flytekit.common.exceptions import system as _system_exceptions
-from flytekit.common.tasks.sdk_runnable import ExecutionParameters
 from flytekit.configuration import TemporaryConfiguration as _TemporaryConfiguration
 from flytekit.configuration import internal as _internal_config
 from flytekit.configuration import sdk as _sdk_config
+from flytekit.core import constants as _constants
+from flytekit.core import utils
 from flytekit.core.base_task import IgnoreOutputs, PythonTask
 from flytekit.core.context_manager import (
+    ExecutionParameters,
     ExecutionState,
     FlyteContext,
     FlyteContextManager,
@@ -33,9 +27,8 @@
 from flytekit.core.data_persistence import FileAccessProvider
 from flytekit.core.map_task import MapPythonTask
 from flytekit.core.promise import VoidPromise
-from flytekit.engines import loader as _engine_loader
-from flytekit.interfaces import random as _flyte_random
-from flytekit.interfaces.data import data_proxy as _data_proxy
+from flytekit.exceptions import scopes as _scoped_exceptions
+from flytekit.exceptions import scopes as _scopes
 from flytekit.interfaces.stats.taggable import get_stats as _get_stats
 from flytekit.loggers import entrypoint_logger as logger
 from flytekit.models import dynamic_job as _dynamic_job
@@ -47,6 +40,12 @@
 from flytekit.tools.module_loader import load_object_from_module
 
 
+def get_version_message():
+    import flytekit
+
+    return f"Welcome to Flyte! Version: {flytekit.__version__}"
+
+
 def _compute_array_job_index():
     # type () -> int
     """
@@ -61,25 +60,6 @@ def _compute_array_job_index():
     return offset + int(_os.environ.get(_os.environ.get("BATCH_JOB_ARRAY_INDEX_VAR_NAME")))
 
 
-def _map_job_index_to_child_index(local_input_dir, datadir, index):
-    local_lookup_file = local_input_dir.get_named_tempfile("indexlookup.pb")
-    idx_lookup_file = _os.path.join(datadir, "indexlookup.pb")
-
-    # if the indexlookup.pb does not exist, then just return the index
-    if not _data_proxy.Data.data_exists(idx_lookup_file):
-        return index
-
-    _data_proxy.Data.get_data(idx_lookup_file, local_lookup_file)
-    mapping_proto = _utils.load_proto_from_file(_literals_pb2.LiteralCollection, local_lookup_file)
-    if len(mapping_proto.literals) < index:
-        raise _system_exceptions.FlyteSystemAssertion(
-            "dynamic task index lookup array size: {} is smaller than lookup index {}".format(
-                len(mapping_proto.literals), index
-            )
-        )
-    return mapping_proto.literals[index].scalar.primitive.integer
-
-
 def _dispatch_execute(
     ctx: FlyteContext,
     task_def: PythonTask,
@@ -101,7 +81,7 @@ def _dispatch_execute(
         # Step1
         local_inputs_file = _os.path.join(ctx.execution_state.working_dir, "inputs.pb")
         ctx.file_access.get_data(inputs_path, local_inputs_file)
-        input_proto = _utils.load_proto_from_file(_literals_pb2.LiteralMap, local_inputs_file)
+        input_proto = utils.load_proto_from_file(_literals_pb2.LiteralMap, local_inputs_file)
         idl_input_literals = _literal_models.LiteralMap.from_flyte_idl(input_proto)
 
         # Step2
@@ -174,7 +154,7 @@ def _dispatch_execute(
         logger.error("!! End Error Captured by Flyte !!")
 
     for k, v in output_file_dict.items():
-        _common_utils.write_proto_to_file(v.to_flyte_idl(), _os.path.join(ctx.execution_state.engine_dir, k))
+        utils.write_proto_to_file(v.to_flyte_idl(), _os.path.join(ctx.execution_state.engine_dir, k))
 
     ctx.file_access.put_data(ctx.execution_state.engine_dir, output_prefix, is_multipart=True)
     logger.info(f"Engine folder written successfully to the output prefix {output_prefix}")
@@ -283,45 +263,6 @@ def _handle_annotated_task(
     _dispatch_execute(ctx, task_def, inputs, output_prefix)
 
 
-@_scopes.system_entry_point
-def _legacy_execute_task(task_module, task_name, inputs, output_prefix, raw_output_data_prefix, test):
-    """
-    This function should be called for old flytekit api tasks (the only API that was available in 0.15.x and earlier)
-    """
-    with _TemporaryConfiguration(_internal_config.CONFIGURATION_PATH.get()):
-        with _utils.AutoDeletingTempDir("input_dir") as input_dir:
-            # Load user code
-            task_module = _importlib.import_module(task_module)
-            task_def = getattr(task_module, task_name)
-
-            local_inputs_file = input_dir.get_named_tempfile("inputs.pb")
-
-            # Handle inputs/outputs for array job.
-            if _os.environ.get("BATCH_JOB_ARRAY_INDEX_VAR_NAME"):
-                job_index = _compute_array_job_index()
-
-                # TODO: Perhaps remove.  This is a workaround to an issue we perceived with limited entropy in
-                # TODO: AWS batch array jobs.
-                _flyte_random.seed_flyte_random(
-                    "{} {} {}".format(_random.random(), _datetime.datetime.utcnow(), job_index)
-                )
-
-                # If an ArrayTask is discoverable, the original job index may be different than the one specified in
-                # the environment variable. Look up the correct input/outputs in the index lookup mapping file.
-                job_index = _map_job_index_to_child_index(input_dir, inputs, job_index)
-
-                inputs = _os.path.join(inputs, str(job_index), "inputs.pb")
-                output_prefix = _os.path.join(output_prefix, str(job_index))
-
-            _data_proxy.Data.get_data(inputs, local_inputs_file)
-            input_proto = _utils.load_proto_from_file(_literals_pb2.LiteralMap, local_inputs_file)
-
-            _engine_loader.get_engine().get_task(task_def).execute(
-                _literal_models.LiteralMap.from_flyte_idl(input_proto),
-                context={"output_prefix": output_prefix, "raw_output_data_prefix": raw_output_data_prefix},
-            )
-
-
 @_scopes.system_entry_point
 def _execute_task(
     inputs,
@@ -416,8 +357,6 @@ def _pass_through():
 
 
 @_pass_through.command("pyflyte-execute")
-@_click.option("--task-module", required=False)
-@_click.option("--task-name", required=False)
 @_click.option("--inputs", required=True)
 @_click.option("--output-prefix", required=True)
 @_click.option("--raw-output-data-prefix", required=False)
@@ -431,8 +370,6 @@ def _pass_through():
     nargs=-1,
 )
 def execute_task_cmd(
-    task_module,
-    task_name,
     inputs,
     output_prefix,
     raw_output_data_prefix,
@@ -442,7 +379,7 @@ def execute_task_cmd(
     resolver,
     resolver_args,
 ):
-    logger.info(_utils.get_version_message())
+    logger.info(get_version_message())
     # We get weird errors if there are no click echo messages at all, so emit an empty string so that unit tests pass.
     _click.echo("")
     # Backwards compatibility - if Propeller hasn't filled this in, then it'll come through here as the original
@@ -455,21 +392,17 @@ def execute_task_cmd(
     # Use the presence of the resolver to differentiate between old API tasks and new API tasks
     # The addition of a new top-level command seemed out of scope at the time of this writing to pursue given how
     # pervasive this top level command already (plugins mostly).
-    if not resolver:
-        logger.info("No resolver found, assuming legacy API task...")
-        _legacy_execute_task(task_module, task_name, inputs, output_prefix, raw_output_data_prefix, test)
-    else:
-        logger.debug(f"Running task execution with resolver {resolver}...")
-        _execute_task(
-            inputs,
-            output_prefix,
-            raw_output_data_prefix,
-            test,
-            resolver,
-            resolver_args,
-            dynamic_addl_distro,
-            dynamic_dest_dir,
-        )
+    logger.debug(f"Running task execution with resolver {resolver}...")
+    _execute_task(
+        inputs,
+        output_prefix,
+        raw_output_data_prefix,
+        test,
+        resolver,
+        resolver_args,
+        dynamic_addl_distro,
+        dynamic_dest_dir,
+    )
 
 
 @_pass_through.command("pyflyte-fast-execute")
@@ -528,7 +461,7 @@ def map_execute_task_cmd(
     resolver,
     resolver_args,
 ):
-    logger.info(_utils.get_version_message())
+    logger.info(get_version_message())
 
     _execute_map_task(
         inputs,

flytekit/clients/friendly.py

@@ -560,7 +560,7 @@ def create_execution(self, project, domain, name, execution_spec, inputs):
     def recover_execution(self, id, name: str = None):
         """
         Recreates a previously-run workflow execution that will only start executing from the last known failure point.
-        :param flytekit.common.core.identifier.WorkflowExecutionIdentifier id:
+        :param flytekit.models.core.identifier.WorkflowExecutionIdentifier id:
         :param name str: Optional name to assign to the newly created execution.
         :rtype: flytekit.models.core.identifier.WorkflowExecutionIdentifier
         """
@@ -572,7 +572,7 @@ def recover_execution(self, id, name: str = None):
 
     def get_execution(self, id):
         """
-        :param flytekit.common.core.identifier.WorkflowExecutionIdentifier id:
+        :param flytekit.models.core.identifier.WorkflowExecutionIdentifier id:
         :rtype: flytekit.models.execution.Execution
         """
         return _execution.Execution.from_flyte_idl(
@@ -638,7 +638,7 @@ def list_executions_paginated(self, project, domain, limit=100, token=None, filt
 
     def terminate_execution(self, id, cause):
         """
-        :param flytekit.common.core.identifier.WorkflowExecutionIdentifier id:
+        :param flytekit.models.core.identifier.WorkflowExecutionIdentifier id:
         :param Text cause:
         """
         super(SynchronousFlyteClient, self).terminate_execution(
@@ -647,7 +647,7 @@ def terminate_execution(self, id, cause):
 
     def relaunch_execution(self, id, name=None):
         """
-        :param flytekit.common.core.identifier.WorkflowExecutionIdentifier id:
+        :param flytekit.models.core.identifier.WorkflowExecutionIdentifier id:
         :param Text name: [Optional] name for the new execution. If not specified, a randomly generated name will be
             used
         :returns: The unique identifier for the new execution.

flytekit/clients/helpers.py

@@ -9,8 +9,8 @@ def iterate_node_executions(
     """
     This returns a generator for node executions.
     :param flytekit.clients.friendly.SynchronousFlyteClient client:
-    :param flytekit.common.core.identifier.WorkflowExecutionIdentifier workflow_execution_identifier:
-    :param flytekit.common.core.identifier.TaskExecutionIdentifier task_execution_identifier:
+    :param flytekit.models.core.identifier.WorkflowExecutionIdentifier workflow_execution_identifier:
+    :param flytekit.models.core.identifier.TaskExecutionIdentifier task_execution_identifier:
     :param int limit: The maximum number of elements to retrieve
     :param list[flytekit.models.filters.Filter] filters:
     :rtype: Iterator[flytekit.models.node_execution.NodeExecution]

flytekit/clients/raw.py

@@ -2,7 +2,6 @@
 import time
 from typing import List
 
-import six as _six
 from flyteidl.service import admin_pb2_grpc as _admin_service
 from google.protobuf.json_format import MessageToJson as _MessageToJson
 from grpc import RpcError as _RpcError
@@ -13,13 +12,13 @@
 
 from flytekit.clis.auth import credentials as _credentials_access
 from flytekit.clis.sdk_in_container import basic_auth as _basic_auth
-from flytekit.common.exceptions import user as _user_exceptions
 from flytekit.configuration import creds as _creds_config
 from flytekit.configuration.creds import _DEPRECATED_CLIENT_CREDENTIALS_SCOPE as _DEPRECATED_SCOPE
 from flytekit.configuration.creds import CLIENT_ID as _CLIENT_ID
 from flytekit.configuration.creds import COMMAND as _COMMAND
 from flytekit.configuration.creds import DEPRECATED_OAUTH_SCOPES, SCOPES
 from flytekit.configuration.platform import AUTH as _AUTH
+from flytekit.exceptions import user as _user_exceptions
 from flytekit.loggers import cli_logger
 
 
@@ -91,9 +90,7 @@ def _refresh_credentials_from_command(flyte_client):
         output = subprocess.run(command, capture_output=True, text=True, check=True)
     except subprocess.CalledProcessError as e:
         cli_logger.error("Failed to generate token from command {}".format(command))
-        raise _user_exceptions.FlyteAuthenticationException(
-            "Problems refreshing token with command: " + _six.text_type(e)
-        )
+        raise _user_exceptions.FlyteAuthenticationException("Problems refreshing token with command: " + str(e))
     flyte_client.set_access_token(output.stdout.strip())
 
 
@@ -134,7 +131,7 @@ def handler(*args, **kwargs):
                         # Always retry auth errors.
                         if i == (max_retries - 1):
                             # Exit the loop and wrap the authentication error.
-                            raise _user_exceptions.FlyteAuthenticationException(_six.text_type(e))
+                            raise _user_exceptions.FlyteAuthenticationException(str(e))
                         cli_logger.error(f"Unauthenticated RPC error {e}, refreshing credentials and retrying\n")
                         refresh_handler_fn = _get_refresh_handler(_creds_config.AUTH_MODE.get())
                         refresh_handler_fn(args[0])