Init

Signed-off-by: Kevin Su <pingsutw@apache.org>

Author: pingsutw

.github/workflows/migration.yml

@@ -0,0 +1,50 @@
+name: Build
+
+# Schedule runs to run twice a day
+
+on:
+  push:
+    branches:
+      - master
+      - 'release-v**'
+  pull_request:
+
+env:
+  FLYTE_SDK_LOGGING_LEVEL: 10 # debug
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number }}
+  cancel-in-progress: true
+
+jobs:
+  build:
+    needs:
+    runs-on: ${{ matrix.os }}
+    strategy:
+      fail-fast: false
+      matrix:
+        os:
+          - ubuntu-24.04-arm
+#          - ubuntu-latest
+#          - windows-latest
+#          - macos-latest
+        python-version:
+          - "3.12"
+    steps:
+      - uses: actions/checkout@v4
+      - name: "Clear action cache"
+        uses: ./.github/actions/clear-action-cache
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v4
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install dependencies
+        run: |
+          pip install uv
+          make setup-global-uv
+          uv pip install --pre flyte
+          uv pip freeze
+      - name: Test migrations
+        run: |
+          python migrate_examples/hello.py
+      

.github/workflows/pythonbuild.yml

@@ -7,7 +7,7 @@ on:
     branches:
       - master
       - 'release-v**'
-  pull_request:
+  # pull_request:
   schedule:
     - cron: "0 13 * * *" # This schedule runs at 1pm UTC every day
 

.gitignore

@@ -40,3 +40,4 @@ coverage.xml
 # Version file is auto-generated by setuptools_scm
 flytekit/_version.py
 testing
+.flyte/

flytekit/__init__.py

@@ -99,6 +99,7 @@
 - BlobType
 
 """
+import flytekit.migrate  # isort: skip
 
 import os
 import sys

flytekit/migrate/__init__.py

@@ -0,0 +1,2 @@
+from flytekit.migrate import task, dynamic, workflow  # noqa: F401
+

flytekit/migrate/dynamic.py

@@ -0,0 +1,13 @@
+from typing import Callable, Union
+
+import flytekit
+
+import flytekit.migrate
+from flyte._task import AsyncFunctionTaskTemplate, P, R
+
+
+def dynamic_shim(**kwargs) -> Union[AsyncFunctionTaskTemplate, Callable[P, R]]:
+    return flytekit.migrate.task.task_shim(**kwargs)
+
+
+flytekit.dynamic = dynamic_shim

flytekit/migrate/task.py

@@ -0,0 +1,84 @@
+import datetime
+from typing import Callable, Dict, Iterable, List, Literal, Optional, Tuple, Union, TYPE_CHECKING
+
+
+import flyte
+from flyte import Image, Resources, TaskEnvironment
+from flyte._doc import Documentation
+from flyte._task import AsyncFunctionTaskTemplate, P, R
+
+import flytekit
+
+if TYPE_CHECKING:
+    from flytekit import Cache, Resources, Secret, ImageSpec, Documentation, PodTemplate
+    from flytekit.core.base_task import T, TaskResolverMixin
+    from flytekit.core.python_function_task import PythonFunctionTask
+    from flytekit.core.task import FuncOut
+    from flytekit.deck import DeckField
+    from flytekit.extras.accelerators import BaseAccelerator
+
+
+def task_shim(
+    _task_function: Optional[Callable[P, "FuncOut"]] = None,
+    task_config: Optional["T"] = None,
+    cache: Union[bool, "Cache"] = False,
+    retries: int = 0,
+    interruptible: Optional[bool] = None,
+    deprecated: str = "",
+    timeout: Union[datetime.timedelta, int] = 0,
+    container_image: Optional[Union[str, "ImageSpec"]] = None,
+    environment: Optional[Dict[str, str]] = None,
+    requests: Optional[Resources] = None,
+    limits: Optional[Resources] = None,
+    secret_requests: Optional[List["Secret"]] = None,
+    docs: Optional["Documentation"] = None,
+    disable_deck: Optional[bool] = None,
+    enable_deck: Optional[bool] = None,
+    pod_template: Optional["PodTemplate"] = None,
+    pod_template_name: Optional[str] = None,
+    accelerator: Optional["BaseAccelerator"] = None,
+    pickle_untyped: bool = False,
+    shared_memory: Optional[Union[Literal[True], str]] = None,
+    resources: Optional[Resources] = None,
+    labels: Optional[dict[str, str]] = None,
+    annotations: Optional[dict[str, str]] = None,
+    **kwargs,
+) -> Union[AsyncFunctionTaskTemplate, Callable[P, R]]:
+    plugin_config = task_config
+    pod_template = (
+        flyte.PodTemplate(
+            pod_spec=pod_template.pod_spec,
+            primary_container_name=pod_template.primary_container_name,
+            labels=pod_template.labels,
+            annotations=pod_template.annotations,
+        )
+        if pod_template
+        else None
+    )
+
+    if isinstance(container_image, flytekit.ImageSpec):
+        image = Image.from_debian_base()
+        if container_image.apt_packages:
+            image = image.with_apt_packages(*container_image.apt_packages)
+        pip_packages = ["flytekit"]
+        if container_image.packages:
+            pip_packages.extend(container_image.packages)
+        image = image.with_pip_packages(*pip_packages)
+    elif isinstance(container_image, str):
+        image = Image.from_base(container_image).with_pip_packages("flyte")
+    else:
+        image = Image.from_debian_base().with_pip_packages("flytekit")
+
+    docs = Documentation(description=docs.short_description) if docs else None
+
+    env = TaskEnvironment(
+        name="flytekit",
+        resources=Resources(cpu=0.8, memory="800Mi"),
+        image=image,
+        cache="enabled" if cache else "disable",
+        plugin_config=plugin_config,
+    )
+    return env.task(retries=retries, pod_template=pod_template_name or pod_template, docs=docs)
+
+
+flytekit.task = task_shim

flytekit/migrate/workflow.py

@@ -0,0 +1,13 @@
+import flytekit
+
+from flyte import Image, Resources, TaskEnvironment
+
+env = TaskEnvironment(
+    name="flytekit",
+    resources=Resources(cpu=0.8, memory="800Mi"),
+    image=Image.from_debian_base().with_apt_packages("vim").with_pip_packages("flytekit", "pandas"),
+)
+
+# TODO: Build subtask's image
+
+flytekit.workflow = env.task

flytekit/remote/remote.py

@@ -12,6 +12,7 @@
 import functools
 import gzip
 import hashlib
+import logging
 import os
 import pathlib
 import tempfile
@@ -1932,6 +1933,12 @@ def execute(
             )
         raise NotImplementedError(f"entity type {type(entity)} not recognized for execution")
 
+    def execute_v2(self, entity, **kwargs):
+        import flyte
+        flyte.init_from_config()
+        run = flyte.with_runcontext(log_level=logging.DEBUG).run(entity, **kwargs)
+        print(run.url)
+
     # Flyte Remote Entities
     # ---------------------
 

migrate_exmaples/hello.py

@@ -0,0 +1,29 @@
+import logging
+
+from flytekit import ImageSpec, dynamic, task, workflow
+
+image = ImageSpec(apt_packages=["vim"], packages=["pandas"])
+
+
+@task(cache=True, cache_version="1.0", retries=3, container_image=image)
+def say_hello(name: str):
+    print(f"Hello, {name}!")
+
+
+@dynamic(container_image=image)
+def dynamic_task(name: str):
+    say_hello(name=name)
+
+
+@workflow
+def wf(name: str):
+    say_hello(name=name)
+    dynamic_task(name=name)
+
+
+if __name__ == "__main__":
+    import flyte
+    flyte.init_from_config(log_level=logging.DEBUG)
+    # run = flyte.with_runcontext(log_level=logging.DEBUG).run(wf, name="flyte")
+    # print(run.name)
+    # print(run.url)