Summary
This article is useful for users who want to take advantage of Parameters, which are passed as arguments to our flow function. Parameters enable us to pass data to a flow that can vary by use case without having to adjust the flow itself.
Video Link
What is Staying the Same
Parameters are available in Prefect 1.0 and 2.0. In both cases, they can be used to set default values which are passed to the flow, as well as custom parameters which can be configured at run time.
What is Changing
In Cloud 1.0, Parameters are implemented in our flow as specialized tasks. In 2.0, flows have first-class support for parameterization. Prefect 2.0 also offers parameter type conversion, which enables Prefect flows and tasks to perform type conversion for the parameters passed to the flow function. In addition, Prefect 2.0 enables our flows to coerce any pydantic model type hint into the correct form automatically at runtime.
How do we do this in Prefect 1.0?
Here is an example of how Parameters are used in 1.0.
from prefect import task, Flow, Parameter
@task
def print_plus_one(x):
print(x + 1)
with Flow("param-flow") as flow:
x = Parameter('x', default = 2)
print_plus_one(x=x)
flow.run(parameters=dict(x=42)) # prints 43
flow.run(parameters=dict(x=100)) # prints 101
flow.run() #use default value of 2, prints 3
This is a very simple flow which adds 1 to the integer that is passed into the flow as an argument. Note how we are importing the Parameter task at the start of the flow in order to use this feature. The default value for the parameter is set to 2, which is used whenever a value is not specified for the parameter. The flow above can be run successfully, without issue.
Now let’s break this flow for the sake of this demo.
from prefect import task, Flow, Parameter
@task
def print_plus_one(x):
print(x + 1)
with Flow("param-flow-fails") as flow:
x = Parameter('x', default = 2)
print_plus_one(x=x)
flow.run(parameters=dict(x="42")) #fails
flow.run(parameters=dict(x="100")) #fails
Notice how we’ve changed the integers 42 and 100 to strings. If we run this flow, it will fail and raise a TypeError. This is something that we can remedy quite easily with our 2.0 flow example.
How do we do this in Prefect 2.0?
Here is an example of how parameters are used in 2.0. Take note that we no longer need to import Parameters in order to use them.
from prefect import task, flow
@task
def print_plus_one(obj):
print(f"Received a {type(obj)} with value {obj}") #Shows the type of the parameter after coercion
print(obj + 1) #Adds one
#Note that we define the flow with type hints
@flow
def validation_flow(x: int, y: int):
print_plus_one(x)
print_plus_one(y)
validation_flow(x="42", y=100)
#The above prints the following:
# Received a <class 'int'> with value 42
# 43
# Received a <class 'int'> with value 100
# 101
This flow executes the same process as our 1.0 example, but it is more robust with handling parameters. We have specified in our flow that both x and y should be coerced into integers. When our flow is run, “42” is passed as a string, and 100 is passed as an integer. The flow does not fail despite being passed a string, and the print_plus_one task runs successfully.
In the output, we can also see that the task receives two arguments with a class of ‘int’, showing that the string is coerced into an integer.