https://prefect.io logo
Join Slack
Channels
Powered by
Hi, I am trying to create a flow based on some dyn...
# ask-community
s
Hi, I am trying to create a flow based on some dynamic values. These dynamic values tell the tasks which are required to create that flow. My tasks are saved in separate files. For eg: task1.py
Copy code
def task1():
    @task
    def run_terraform_lt():
        tf = Terraform(working_dir="law_tf")
        tf.init()
		tf.apply()
task2.py
Copy code
def task2():
    @task
    def run_terraform_rt():
        tf = Terraform(working_dir="rt_tf")
        tf.init()
		tf.apply()
The input is similar to this:
Copy code
{
    "task1": {
      "id": "foo",
      "task_name": "task1"
    },
    "task2": {
      "id": "bar",
      "task_name": "task2",
    }
}
I am getting the task name from this dictionary. Based on the value of "task_name", I have to create a flow combining all tasks. I am creating a flow where I am trying to add them, but the tasks are not getting added in the flow. Anyone has any idea on how this can be achieved? Here is the snippet of my flow creation code. It is inside a for loop through which I am extracting the task name and other variables.
k
I think the first thing here is the tasks not being added to the flow. Why is your syntax like that?
Copy code
def task2():
    @task
    def run_terraform_rt():
        tf = Terraform(working_dir="rt_tf")
        tf.init()
		tf.apply()
How do you call this in the Flow?
s
i was thinking if i can somehow call task2(), that may work. but if you have any other suggestion, I can try it
k
But you can already do:
Copy code
@task
def run_terraform_rt():
    tf = Terraform(working_dir="rt_tf")
    tf.init()
    tf.apply()
and then call run_terraform_rt in the flow. Why do you need the outer function?
s
since the tasks name (in turn the filename) is dynamic, normal import was not working. i had to use importlib and then there were some errors without the outer function. but now i'll try once again i cannot directly call run_terraform_rt() because the task list is dynamic. this is where the problem is
k
I think you are thinking to dynamically add tasks to the Flow. That is not that way to try this because the static DAG has to be created during registration time. Instead, what you need to do is make the full DAG and register it, and then conditionally skip tasks based on the parameters.
Or you can put more logic inside the task to handle the logic based on the parameter inputs. Can you use if-else inside a task to describe your logic?
s
Yes I can use if-else inside the task
This is the msg I am getting now
I have not yet used if-else. This is just the outer function removed
k
Can I see the code?
s
Copy code
@task
def task1():
    tf = Terraform(working_dir="tf")
    tf.init()
    tf.plan()
    tf.apply(skip_plan=True)
k
I think your code is doing something like this:
Copy code
from prefect import Flow, task

with Flow("name") as flow:
    @task
    def abc():
        return 1
which will complain because the task is defined but not called. It should really be defined outside then used
Copy code
@task
def abc():
    return 1
with Flow("name") as flow:
    abc()
or you could all it
Copy code
from prefect import Flow, task

with Flow("name") as flow:
    @task
    def abc():
        return 1
    abc()
but i don’t think we’d recommend defining the tasks inside the Flow
s
I understand your point. I just want to call the task which is defined in a different file in my flow. the only problem is that the task name is not static, so I cannot use a static name. I have to get it from the input which is passed as a dictionary
k
This is to build a flow, not run right?
s
yes. just to build
k
Maybe you can use the Python 
exec
function which takes in a string and runs it as python code? but I think this will not work well during run time. I think the Flow may have problems finding that task definition. I dont know
s
ok, i'll try that. right now i am using exec_module
k
Wait though, I am wondering why this can’t go inside the task?
Copy code
@task
def task_one(val):
    if val == "func_a":
        func_a()
    if val == "func_b":
        func_b()
like that?
s
we are planning to have a lot of tasks which perform a single activity. then based on some request (a json request), a flow needs to be created which will consist of mix and match of those tasks. The tasks are creating a single resource in Azure. The json request can be to create any resource and any number of resource. One request can ask to create resource group and virtual machine. another one can ask for security group and database. etc. So we can't hardcode which tasks will go in the flow.
k
I understand more, but I can’t picture how to do it elegantly. Let me just say that you can build the DAG conditionally as well:
Copy code
def create_flow(var):
    with Flow(...) as flow:
        if var:
            x = task_one()
        else:
            x = task_two()
        task_two(x)
I think this might work
I encourage you though to make a smaller proof of concept because I don’t know if it will load the tasks during runtime properly. I think it might.
s
ok. I'll keep trying. I'll post the answer here if I find anything
58 Views
Open in Slack
PreviousNext
Powered by