Great question. In Prefect 3.x, the most common team layout follows a standard Python “src” layout with a dedicated flows directory, a versioned prefect.yaml at the repo root, and CI/CD + infra alongside. Here’s a solid, widely used structure:

your-project/
├─ pyproject.toml
├─ prefect.yaml
├─ README.md
├─ .gitignore
├─ .env.example
├─ flows/                      # Flow entrypoints (thin wrappers that call shared code)
│  ├─ etl_daily.py             # e.g., contains `@flow` functions
│  └─ backfill_orders.py
├─ src/your_project/           # Reusable code (tasks, utils, domain logic)
│  ├─ __init__.py
│  ├─ tasks/
│  │  ├─ io.py
│  │  └─ transforms.py
│  ├─ blocks.py                # Optional: custom Blocks defined in Python
│  └─ utils.py
├─ tests/                      # pytest tests for flows and shared code
│  ├─ test_flows.py
│  └─ test_tasks.py
├─ infra/                      # Infra-as-code used by deployments/workers
│  ├─ Dockerfile
│  ├─ kubernetes/
│  │  └─ job-template.json     # if using K8s work pools with custom job templates
│  └─ terraform/
│     └─ ...                   # if managing Prefect objects/infra via Terraform
└─ .github/workflows/
   └─ ci.yml                   # Lint/test; optionally run `prefect deploy` in CI

Key practices behind this layout: - Keep flow files in

flows/

as thin entrypoints and put reusable logic under

src/your_project/

. - Version-control a single

prefect.yaml

at the repo root that defines deployments (and build/push/pull steps if you need them). - If you define custom Blocks, keep them in code (e.g.,

src/your_project/blocks.py

) and register them with:

prefect block register -f src/your_project/blocks.py

- Use CI to test and update deployments via:

prefect deploy

- Prefer

flow.from_source(...)

to load flows from your Git repo when you build deployments for remote execution (instead of legacy GitHub storage blocks). A minimal

prefect.yaml

example for a repo like the above:

deployments:
  - name: etl-daily
    entrypoint: flows/etl_daily.py:etl_daily
    work_pool:
      name: prod-k8s
    parameters:
      start_date: "{{ $START_DATE | default('2024-01-01') }}"
    pull:
      - prefect.deployments.steps.set_working_directory:
          directory: "."
      - prefect.deployments.steps.git_clone:
          repository: <https://github.com/your-org/your-project.git>
          branch: main
      - prefect.deployments.steps.pip_install_requirements:
          requirements_file: pyproject.toml

Handy commands we verified: - Initialize a project with a recipe (creates

prefect.yaml

and scaffolding):

prefect init

- Create or update deployments from your project:

prefect deploy

- Register custom block types defined in your code:

prefect block register -f src/your_project/blocks.py

References: - Create deployments - prefect.yaml reference - Store and load flow code from source (flow.from_source) - Custom Blocks If you’re in a monorepo, the same pattern applies; either: - one

prefect.yaml

at the root with multiple deployments using different entrypoints, or - separate service folders each with their own

prefect.yaml

and flows. If you share more about your environment (monorepo vs multi-repo, Docker/K8s, CI system), I can tailor this to your setup.

Absolutely—here’s a common team layout that favors Python-based deployments (no prefect.yaml), while keeping the standard “src layout” and thin flow entrypoints. Recommended repo structure

your-project/
├─ pyproject.toml
├─ README.md
├─ .gitignore
├─ .env.example
├─ flows/                      # Thin flow entrypoints
│  ├─ etl_daily.py
│  └─ backfill_orders.py
├─ src/your_project/           # Reusable code (tasks, utils, domain logic)
│  ├─ __init__.py
│  ├─ tasks/
│  │  ├─ io.py
│  │  └─ transforms.py
│  ├─ blocks.py                # Optional: custom Blocks defined in Python
│  └─ utils.py
├─ tests/
│  ├─ test_flows.py
│  └─ test_tasks.py
├─ deploy/                     # Python deployment scripts (replaces prefect.yaml)
│  └─ deploy_all.py
├─ infra/
│  ├─ Dockerfile
│  ├─ kubernetes/
│  │  └─ job-template.json     # if using K8s work pools with custom job templates
│  └─ terraform/
│     └─ ...
└─ .github/workflows/
   └─ ci.yml                   # Lint/test; run Python deploy script(s) in CI

Example flow (thin entrypoint)

# flows/etl_daily.py
from prefect import flow, task
from prefect.logging import get_run_logger

@task
def extract():
    return [1, 2, 3]

@task
def transform(xs):
    return [x * 2 for x in xs]

@task
def load(xs):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Loaded: {xs}")

@flow
def etl_daily(start_date: str = "2024-01-01"):
    data = extract()
    data = transform(data)
    load(data)

Python-only deployments (no YAML) - Use Flow.deploy for each deployment. - Use Flow.from_source so workers can fetch your flow code from Git at run time. - Pass schedules, tags, parameters, and infra settings directly in Python.

# deploy/deploy_all.py
import os
from prefect.docker import DockerImage
from flows.etl_daily import etl_daily
from flows.backfill_orders import backfill_orders

# Build/push config (works with Prefect’s build pipeline when deploying)
IMAGE = DockerImage(
    name="<http://ghcr.io/your-org/your-project|ghcr.io/your-org/your-project>",
    tag=os.getenv("GIT_SHA", "latest"),
    dockerfile="infra/Dockerfile",  # "auto" also works if you prefer
)

# Prefer from_source so workers can fetch code at runtime
etl_daily_src = etl_daily.from_source(
    source="<https://github.com/your-org/your-project.git>",
    entrypoint="flows/etl_daily.py:etl_daily",
)

etl_daily_src.deploy(
    name="etl-daily",
    work_pool_name="prod-k8s",      # or your process/docker/k8s pool name
    image=IMAGE,                    # triggers build/push by default
    cron="0 2 * * *",               # or interval/rrule/schedule objects
    parameters={"start_date": "2024-01-01"},
    tags=["etl", "daily"],
    # For K8s work pools, pass base job template fields here if needed:
    job_variables={"image_pull_policy": "IfNotPresent"},
)

backfill_src = backfill_orders.from_source(
    source="<https://github.com/your-org/your-project.git>",
    entrypoint="flows/backfill_orders.py:backfill_orders",
)

backfill_src.deploy(
    name="backfill-orders",
    work_pool_name="prod-k8s",
    image=IMAGE,
    paused=True,                    # keep paused; run manually or via triggers
)

How to run deployment updates - Locally or in CI, just run your deploy script:

python -m deploy.deploy_all

- Ensure

PREFECT_API_URL

and

PREFECT_API_KEY

are set in your environment. Blocks and secrets in Python - Define custom Blocks in code (e.g.,

src/your_project/blocks.py

) and register the type once:

prefect block register -f src/your_project/blocks.py

- Load them in flows at runtime:

from prefect.blocks.system import Secret
api_key = Secret.load("my-service-api-key").get()

Notes and tips - Keep flows in

flows/

as thin entrypoints; put reusable logic in

src/your_project/

. - For bulk deployments, keep everything in

deploy/deploy_all.py

and call

.deploy

for each flow you want to publish. - If you don’t use containers (e.g., process workers), omit

image=...

, but still prefer

from_source(...)

so workers can pull the code.

- For K8s/Docker workers, prefer specifying an image so runs are reproducible. Docs and references - Create deployments with Python, including

Flow.deploy

and `flow.from_source`: - Create deployments - Store/load flow code from source - Schedules (Cron/Interval/RRule) are accepted as kwargs to

Flow.deploy

and also available as objects: - Deployments and schedules - Custom Blocks: - Custom Blocks If you can share your worker type (process, Docker, or K8s) and whether you’re in a monorepo, I can tailor the deploy script and job_variables for your setup.