Hi, I’m wondering what the best practice with regards to use common function across multiple flows:

utils.py

def common(a):
    # imagine a function that makes over network calls 
    pass

Option 1: create a common task that’s then invoked from the flows like so:

prefect_common.py

from prefect import task

from utils import common

@task
def common_task(a):
    return common(a)

flow_a.py

from prefect import task

from prefect_common import common_task

@task
def task_one():
    pass

@flow
def flow():
    common_task("foo")
    task_one()

flow_b.py

from prefect import task

from prefect_common import common_task

@task
def task_two():
    pass

@flow
def flow():
    common_task("bar")
    task_two()

Option 2: create separate task in each flow that uses the common function

flow_a.py

from prefect import task

from util import common

@task
def task_one():
    pass

@task
def common_task(a):
    return common(a)

@flow
def flow():
    common_task("foo")
    task_one()

flow_b.py

from prefect import task

from util import common

@task
def task_two():
    pass

@task
def common_task(a):
    return common(a)

@flow
def flow():
    common_task(5)
    task_two()

Option 1 seems like the pythonic choice (DRY etc) and Marvin suggested that, but my concern with that is losing configurability of the task specific to each flow (different retry and timeout options, name, etc.)

👋 hey, you're able to configure those task specific options dynamically if you'd like using

.with_options()

https://github.com/PrefectHQ/prefect/blob/main/src/prefect/tasks.py#L726-L762

Excellent, thanks @Jake Kaplan!