flow_decorator.py

from prefect import task, Flow, Parameter


@task
def print_plus_one(x):
    print(x + 1)


def flow(name):
    def inner(fn):
        with Flow(name) as f:
            fn()
        return f

    return inner


@flow(name="Parameterized Flow")
def f():
    x = Parameter('x', default=2)
    print_plus_one(x=x)


# with Flow('Parameterized Flow') as f:
#     x = Parameter('x', default=2)
#     print_plus_one(x=x)

f.run(parameters=dict(x=1))  # prints 2
f.run(parameters=dict(x=100))  # prints 101
f.run()  # prints 3