Hi team. I'm new to Flyte and had a question about decorating functions as tasks. Suppose I have all my functions in their own file (

my_functions.py

) and I cannot change it. That is, I can't go into it and decorate the functions with

@task

. Presently I've got a file that imports the necessary functions, creates tasks, and then builds an

ImperativeWorkflow

. The tasks and workflows are created as shown in `my_workflow.py`:

from flytekit import task
from flytekit.core.workflow import ImperativeWorkflow
from my_functions import func1

func1 = task(func1)

workflow = ImperativeWorkflow(name="my_workflow")
workflow.add_task(func1)

However, when I try to run with this, it appears that Flyte is going and finding

my_functions.func1

and trying to run it directly. Except that it isn't a

PythonTask

object since in the main source code

func1

is not decorated and it just grabs the actual function itself. Is there a way to handle this? Can I create Tasks from functions in code I can't modify/add

@task

decorators to?

You can create alias tasks for each of those in your file or you will have to write a new taskresolver

Thanks for you reply! I can take a look at writing my own resolver. What is the process for creating an alias task?

I mean an object that looks like your function in your module

That's what I thought I was doing by creating

func1

inside

my_workflow.py

as a

task

wrapped version of

func1

imported from

my_functions.py

. However the default resolver finds

my_functions.func1

instead of

my_workflow.func1

. After looking at the resolver, it looks like I could hack together my own resolver that addresses this, but I feel like the better solution is the task aliasing -- if there's a good way to do that. I'm interested to see what you're able to figure out.

No think just that I couldn’t try

@Ketan (kumare3) Any updates?

@Brandon Harrison sorry for the delay, this is what i did

from orig import hello
from flytekit import task, workflow

hello_task = task(hello, cache=True, cache_version="1")


@workflow
def wf() -> str:
    return hello_task(s="world")


if __name__ == "__main__":
    wf()

in orig.py

def hello(s: str) -> str:
    return f"hello {s}"

Thanks for looking into this for me! The steps you took were the same I originally tried and found the same errors you did. I wrote my own resolver that is probably not the best solution, but does get me what I need for right now. The other option of modifying the imported code on the fly in concert with the personal task resolver is probably the most robust solution.

I agree, we should make it work