Hey All I have a case where it would be useful to dynamicall

Question

Hey All I have a case where it would be useful to dynamically pass different functions to a flyte workflow but running into issues when trying to run it Minimal code examples inside thanks

Victor Churikov · Answer

This fails at runtime but registers successfully ```def sometask inputs dict for item in inputs print f Found item class name item decorated task = task sometask workflow def generic wf inputs dict decorated task inputs=inputs ``` This works ``` task def sometask inputs dict for item in inputs print f Found item class name item workflow def generic wf inputs dict sometask inputs=inputs ``` The failing code fails with ```tar Removing leading ` from member names Traceback most recent call last root anaconda3 envs trueself bin pyflyte execute 8 in lt module gt 8 sys exit execute task cmd root anaconda3 envs trueself lib python3 10 site packages click core py 113 0 in call 1130 return self main args kwargs root anaconda3 envs trueself lib python3 10 site packages click core py 105 5 in main 1055 rv = self invoke ctx root anaconda3 envs trueself lib python3 10 site packages click core py 140 4 in invoke 1404 return ctx invoke self callback ctx params root anaconda3 envs trueself lib python3 10 site packages click core py 760 in invoke 760 return callback args kwargs root anaconda3 envs trueself lib python3 10 site packages flytekit bin entr ypoint py 480 in execute task cmd 480 execute task root anaconda3 envs trueself lib python3 10 site packages flytekit exceptio ns scopes py 160 in system entry point 160 return wrapped args kwargs root anaconda3 envs trueself lib python3 10 site packages flytekit bin entr ypoint py 362 in execute task 362 handle annotated task ctx task def inputs output prefix root anaconda3 envs trueself lib python3 10 site packages flytekit bin entr ypoint py 304 in handle annotated task 304 dispatch execute ctx task def inputs output prefix root anaconda3 envs trueself lib python3 10 site packages flytekit bin entr ypoint py 80 in dispatch execute 80 logger debug f Starting dispatch execute for task def name AttributeError function object has no attribute name ```

Victor Churikov · Answer

Found this test in flytekit repo ```def test py func task get container def foo i int pass default img = Image name= default fqn= tag= tag1 other img = Image name= other fqn= tag= tag other cfg = ImageConfig default image=default img images= default img other img settings = SerializationSettings project= p domain= d version= v image config=cfg env= FOO bar pytask = PythonFunctionTask None foo None environment= BAZ baz c = pytask get container settings assert c image == assert c env == FOO bar BAZ baz ``` It seems like I can use PythonFunctionTask to achieve this In the test they use this class to run an undecorated python function or so it seems From my initial effort it seems that it is not recognizing the task config but I need to double check that I am doing it correctly and rely on this example

Pryce · Answer

Interesting it looks like task either adds an attr or calling the decorator directly strips it I imagine there s a way to hack it and get it to work but perhaps the cleaner way would be to use Flyte s builtin conditionals <https docs flyte org projects cookbook en latest auto examples control flow conditions html>

Yee · Answer

there are also dynamic tasks

Yee · Answer

could you elaborate a bit more on the goal

Victor Churikov · Answer

Developers asked me to create a wrapper with Flyte that runs arbitrary code on the cloud in kubernetes It seems like a convenient way to run code because it can be scaled beyond our office machines capacity is tracked in the UI and has various useful features and integrations Few months ago I managed to achieve this with Prefect v2 another ML pipelines solution like Flyte but Prefect did not allow as much flexibility with the scheduling and pod configurations so I ended up looking for a more kubernetes native solution and that s how I found Flyte note this is not our main use case we also run full ML pipelines in addition to this and Flyte has been doing really well so far The developers expect that ideally they should have to mess with Flyte specific concepts as little as possible and are asking me to abstract away executions of single functions on Flyte I d like to achieve this using a ` workflow` or a dynamic rather than a ` task` because that would allow me to add additional nodes to each execution for sending notifications preparing inputs etc But the main task should be something that is dynamically specified by passing a python function to some Flyte wrapper registering it and executing it I currently use FlyteRemote with register script and execute if that s relevant

Victor Churikov · Answer

After some trial and error and looking into it I got this minimal code which successfully registers and executes It receives a dict which I tested with both strings and ints so it can have multiple values of different types as a single input output ```from typing import Any Dict from flytekit import task workflow import functools My own custom helper to make it easier to configure tasks from flyte helpers k8s import task config decorator kwargs = task config cache=True node pool= t4 memory pool image= us central1 docker pkg dev mygcpproject garus1 mycondaimage v1 cpu= 4 memory= 16Gi gpu= 1 def convert to flyte task local function decorator kwargs task decorator kwargs functools wraps local function def flyte task function inputs Dict str Any gt Any return local function inputs return flyte task function def generic task inputs dict gt None print Received inputs inputs generic task = convert to flyte task generic task decorator kwargs workflow def my workflow inputs dict gt None generic task inputs=inputs ``` Now this doesn t fail during runtime because the task and the function behind it have the exact same name I m looking into having the task dynamically named after the function so I can support any function name with this and I ll also test it with actual functions that do work and with real inputs outputs to see if it still works and see if I can get it to support files as well