<@ULVA73B9P> I have a Prefect project with some he...
# ask-marvin
d
@Marvin I have a Prefect project with some helper functions (which are tasks), and all my functions are in one main script. I want to modularize my code and have some of my helper functions in a utils.py file. How would I have that setup, but still be able to call those functions as tasks within my main script?
m
thought for 159.9 seconds
Short answer: put your
@task
-decorated helpers in
utils.py
, import them in your main flow, and call them like normal functions. When called inside a flow, they run as tasks. Example (Prefect 3.x assumed) Project layout (simple):
Copy code
my_project/
├─ main.py
└─ utils.py
utils.py:
Copy code
from prefect import task, get_run_logger

@task(retries=2, retry_delay_seconds=5, name="utils.transform")
def transform(x: int) -> int:
    logger = get_run_logger()
    <http://logger.info|logger.info>("Transforming %s", x)
    return x * 2

@task
def fetch() -> int:
    return 21
main.py:
Copy code
from prefect import flow
from utils import fetch, transform

@flow
def etl():
    value = fetch()
    result = transform(value)
    return result

if __name__ == "__main__":
    print(etl())
That’s it. Import the task functions and call them; Prefect schedules them as task runs because you’re inside a flow. If you want a slightly more scalable layout (package + src layout):
Copy code
my_project/
├─ pyproject.toml
├─ prefect.yaml
└─ src/
   └─ my_project/
      ├─ __init__.py
      ├─ flows/
      │  └─ main_flow.py
      └─ utils.py
src/my_project/utils.py:
Copy code
from prefect import task

@task
def helper(a: int, b: int) -> int:
    return a + b
src/my_project/flows/main_flow.py:
Copy code
from prefect import flow
from my_project.utils import helper

@flow
def run():
    return helper(1, 2)
Notes and gotchas - Circular imports: don’t have
utils.py
import from your flow module. Keep helpers independent. - Tasks inside tasks: calling a task from inside another task executes it like a normal function (no separate task run). If you need orchestration between steps, have the flow call them, or use a subflow. - Naming: give tasks meaningful names (
name="utils.transform"
) so they’re easy to spot in the UI. - Deployments: just make sure your
utils.py
is in the source/image your worker can import. In 3.x, use
flow.from_source(...).deploy(...)
if you’re deploying from a repo, and set your entrypoint to the flow (e.g.,
src/my_project/flows/main_flow.py:run
). Docs, if you enjoy light reading: - Tasks in Prefect 3 - Prefect 3.0 release notes (for context on behavior changes) If you share your current layout, I can tailor the imports/entrypoint so your future self doesn’t curse present you.