Hi all, I'm a newbie to prefect. I have created a flow that uses to

prefect.Parameter

Tasks. The type of the parameters are just native

str

and

int

The flow's logic uses some other tasks that are using the parameter tasks. Those tasks are using some other helper functions that help me reuse code and make the code more readable. f I want that my helper function to use one of the parameter (I just need the value, not the task itself) I need to set it as a prefect task by itself and when calling it from other task it should be called with

.run

since inside the task it isn't in the context of a flow. For example:

import prefect

from prefect import Parameter, Flow, task

with Flow("my flow") as my_flow:

id = Parameter("id", required=True)  # int value

description = Parameter("description", required=True). # str value

result1 = my_task1(id, description)

result2 = my_task2(id, description)

my_reduce_task(result1, result2)

@task

def my_task1(id, description):

val1 = calc_logic_func1()

return shared_task.run(id, val1)

@task

def my_task2(id, description):

val2 = calc_logic_func2()

return shared_task.run(id, val2)

@task

def shared_task(id: int, value:int) -> int:

return ...

def calc_logic_func1() -> int:

return ...

In the example above I want to use a helper function

shared_task

with the integer

id

value but since

id

is a

prefect Parameter Task

, therefore shared_task itself must be a task and calling it from other task (e.g.

my_task1

it should be called with

shared_task.run

. Well, this is how I understand this so far. Is there any other way to use it? (not setting

shared_task

as a task OR not calling it with

.run

since

my_task1

is already called from `my_flow`context) If this is the right way to use it - are there any other effects on the flow run (I guess

my_task1

will execute

shared_task

itself in the same agent) Thanks!

Parameters are usually only needed to inject some runtime-specific value into the DAG structure. Instead of calling its .run() method within another task, it would be better if you would pass it as data dependency between tasks and all tasks that need this as input can use it as input in the Flow structure. You also asked about side effects: one is that you set the Parameter as required without providing default values. This will have the effect that the flow cannot be scheduled, it can only be executed if you execute it ad hoc (manually, via API/CLI) and provide the value this way. We’ve recently published this tutorial about Parameters that explains all that in more detail.

Thanks @Anna Geller for your reply! In the case of moving all the utilities shared tasks to the flow - this will make the flow itself too detailed and somewhat confusing. I'd like to keep the main tasks that together describes the business logic of the flow and just call the "detailed" tasks from those main tasks so I guess using

.run

is the right way in my case? Thanks!

Perhaps you can access parameter value from the context in that case? The syntax is

perfect.context.parameters["parametername"]