Auto Backfill workflow

flytekit/clis/sdk_in_container/backfill.py

@@ -0,0 +1,174 @@
+import typing
+from datetime import datetime, timedelta
+
+import click
+
+from flytekit.clis.sdk_in_container.helpers import get_and_save_remote_with_click_context
+from flytekit.clis.sdk_in_container.run import DateTimeType, DurationParamType
+
+_backfill_help = """
+Backfill command generates, registers a new workflow based on the input launchplan, that can be used to run an
+automated backfill. The workflow can be managed using the Flyte UI and can be canceled, relaunched, recovered and
+is implicitly cached. 
+"""
+
+
+def resolve_backfill_window(
+    from_date: datetime = None,
+    to_date: datetime = None,
+    window: timedelta = None,
+) -> typing.Tuple[datetime, datetime]:
+    """
+    Resolves the from_date -> to_date
+    """
+    if from_date and to_date and window:
+        raise click.BadParameter("Cannot use from-date, to-date and duration. Use any two")
+    if not (from_date or to_date):
+        raise click.BadParameter(
+            "One of following pairs are required -> (from-date, to-date) | (from-date, duration) | (to-date, duration)"
+        )
+    if from_date and to_date:
+        pass
+    elif not window:
+        raise click.BadParameter("One of start-date and end-date are needed with duration")
+    elif from_date:
+        to_date = from_date + window
+    else:
+        from_date = to_date - window
+    return from_date, to_date
+
+
+@click.command("backfill", help=_backfill_help)
+@click.option(
+    "-p",
+    "--project",
+    required=False,
+    type=str,
+    default="flytesnacks",
+    help="Project to register and run this workflow in",
+)
+@click.option(
+    "-d",
+    "--domain",
+    required=False,
+    type=str,
+    default="development",
+    help="Domain to register and run this workflow in",
+)
+@click.option(
+    "-v",
+    "--version",
+    required=False,
+    type=str,
+    default=None,
+    help="Version for the registered workflow. If not specified it is auto-derived using the start and end date",
+)
+@click.option(
+    "-n",
+    "--execution-name",
+    required=False,
+    type=str,
+    default=None,
+    help="Create a named execution for the backfill. This can prevent launching multiple executions.",
+)
+@click.option(
+    "--dry-run",
+    required=False,
+    type=bool,
+    is_flag=True,
+    default=False,
+    show_default=True,
+    help="Just generate the workflow - do not register or execute",
+)
+@click.option(
+    "--parallel/--serial",
+    required=False,
+    type=bool,
+    is_flag=True,
+    default=False,
+    show_default=True,
+    help="All backfill can be run in parallel - with max-parallelism",
+)
+@click.option(
+    "--no-execute",
+    required=False,
+    type=bool,
+    is_flag=True,
+    default=False,
+    show_default=True,
+    help="Generate the workflow and register, do not execute",
+)
+@click.option(
+    "--from-date",
+    required=False,
+    type=DateTimeType(),
+    default=None,
+    help="Date from which the backfill should begin. Start date is inclusive.",
+)
+@click.option(
+    "--to-date",
+    required=False,
+    type=DateTimeType(),
+    default=None,
+    help="Date to which the backfill should run_until. End date is inclusive",
+)
+@click.option(
+    "--backfill-window",
+    required=False,
+    type=DurationParamType(),
+    default=None,
+    help="Timedelta for number of days, minutes hours after the from-date or before the to-date to compute the"
+    " backfills between. This is needed with from-date / to-date. Optional if both from-date and to-date are provided",
+)
+@click.argument(
+    "launchplan",
+    required=True,
+    type=str,
+    # help="Name of launchplan to be backfilled.",
+)
+@click.argument(
+    "launchplan-version",
+    required=False,
+    type=str,
+    default=None,
+    # help="Version of the launchplan to be backfilled, if not specified, the latest version "
+    #      "(by registration time) will be used",
+)
+@click.pass_context
+def backfill(
+    ctx: click.Context,
+    project: str,
+    domain: str,
+    from_date: datetime,
+    to_date: datetime,
+    backfill_window: timedelta,
+    launchplan: str,
+    launchplan_version: str,
+    dry_run: bool,
+    no_execute: bool,
+    parallel: bool,
+    execution_name: str,
+    version: str,
+):
+    from_date, to_date = resolve_backfill_window(from_date, to_date, backfill_window)
+    remote = get_and_save_remote_with_click_context(ctx, project, domain)
+    entity = remote.launch_backfill(
+        project=project,
+        domain=domain,
+        from_date=from_date,
+        to_date=to_date,
+        launchplan=launchplan,
+        launchplan_version=launchplan_version,
+        execution_name=execution_name,
+        version=version,
+        dry_run=dry_run,
+        no_execute=no_execute,
+        parallel=parallel,
+        output=click.secho,
+    )
+    if entity:
+        console_url = remote.generate_console_url(entity)
+        if no_execute:
+            click.secho(f"\n No Execution mode: Workflow registered at {console_url}", fg="green")
+        else:
+            click.secho(f"\n Execution can be seen at {console_url} to see execution in the console.", fg="green")

flytekit/clis/sdk_in_container/pyflyte.py

@@ -1,6 +1,7 @@
 import click
 
 from flytekit import configuration
+from flytekit.clis.sdk_in_container.backfill import backfill
 from flytekit.clis.sdk_in_container.constants import CTX_CONFIG_FILE, CTX_PACKAGES
 from flytekit.clis.sdk_in_container.init import init
 from flytekit.clis.sdk_in_container.local_cache import local_cache
@@ -70,6 +71,7 @@ def main(ctx, pkgs=None, config=None):
 main.add_command(init)
 main.add_command(run)
 main.add_command(register)
+main.add_command(backfill)
 
 if __name__ == "__main__":
     main()

flytekit/clis/sdk_in_container/run.py

@@ -105,12 +105,32 @@ def convert(
         raise click.BadParameter(f"parameter should be a valid file path, {value}")
 
 
+class DateTimeType(click.DateTime):
+
+    _NOW_FMT = "now"
+    _ADDITONAL_FORMATS = [_NOW_FMT]
+
+    def __init__(self):
+        super().__init__()
+        self.formats.extend(self._ADDITONAL_FORMATS)
+
+    def convert(
+        self, value: typing.Any, param: typing.Optional[click.Parameter], ctx: typing.Optional[click.Context]
+    ) -> typing.Any:
+        if value in self._ADDITONAL_FORMATS:
+            if value == self._NOW_FMT:
+                return datetime.datetime.now()
+        return super().convert(value, param, ctx)
+
+
 class DurationParamType(click.ParamType):
-    name = "timedelta"
+    name = "[1:24 | :22 | 1 minute | 10 days | ...]"
 
     def convert(
         self, value: typing.Any, param: typing.Optional[click.Parameter], ctx: typing.Optional[click.Context]
     ) -> typing.Any:
+        if value is None:
+            raise click.BadParameter("None value cannot be converted to a Duration type.")
         return datetime.timedelta(seconds=parse(value))
 
 

flytekit/core/promise.py

@@ -811,6 +811,8 @@ def extract_obj_name(name: str) -> str:
 def create_and_link_node_from_remote(
     ctx: FlyteContext,
     entity: HasFlyteInterface,
+    _inputs_not_allowed: typing.Set[str] = None,
+    _ignorable_inputs: typing.Set[str] = None,
     **kwargs,
 ):
     """
@@ -824,9 +826,18 @@ def create_and_link_node_from_remote(
 
     typed_interface = entity.interface
 
+    if _inputs_not_allowed:
+        inputs_not_allowed_specified = _inputs_not_allowed.intersection(kwargs.keys())
+        if inputs_not_allowed_specified:
+            raise _user_exceptions.FlyteAssertion(
+                f"Fixed inputs cannot be specified. Please remove the following inputs - {inputs_not_allowed_specified}"
+            )
+
     for k in sorted(typed_interface.inputs):
         var = typed_interface.inputs[k]
         if k not in kwargs:
+            if k in _ignorable_inputs or k in _inputs_not_allowed:
+                continue
             # TODO to improve the error message, should we show python equivalent types for var.type?
             raise _user_exceptions.FlyteAssertion("Missing input `{}` type `{}`".format(k, var.type))
         v = kwargs[k]

flytekit/core/workflow.py

@@ -713,7 +713,7 @@ def workflow(
     failure_policy: Optional[WorkflowFailurePolicy] = None,
     interruptible: bool = False,
     docs: Optional[Documentation] = None,
-):
+) -> WorkflowBase:
     """
     This decorator declares a function to be a Flyte workflow. Workflows are declarative entities that construct a DAG
     of tasks using the data flow between tasks.

flytekit/remote/entities.py

@@ -4,9 +4,11 @@
 
 from typing import Dict, List, Optional, Tuple, Union
 
+from flytekit import FlyteContext
 from flytekit.core import constants as _constants
 from flytekit.core import hash as _hash_mixin
 from flytekit.core import hash as hash_mixin
+from flytekit.core.promise import create_and_link_node_from_remote
 from flytekit.exceptions import system as _system_exceptions
 from flytekit.exceptions import user as _user_exceptions
 from flytekit.loggers import remote_logger
@@ -798,5 +800,16 @@ def resource_type(self) -> id_models.ResourceType:
     def entity_type_text(self) -> str:
         return "Launch Plan"
 
+    def compile(self, ctx: FlyteContext, *args, **kwargs):
+        fixed_input_lits = self.fixed_inputs.literals or {}
+        default_input_params = self.default_inputs.parameters or {}
+        return create_and_link_node_from_remote(
+            ctx,
+            entity=self,
+            _inputs_not_allowed=set(fixed_input_lits.keys()),
+            _ignorable_inputs=set(default_input_params.keys()),
+            **kwargs,
+        )  # noqa
+
     def __repr__(self) -> str:
         return f"FlyteLaunchPlan(ID: {self.id} Interface: {self.interface}) - Spec {super().__repr__()})"