Great question. Here’s a pragmatic, Prefect 3–first blueprint we recommend for maintainability and frictionless deployment to Prefect Cloud’s Serverless Compute. Assumptions: You’re on Prefect 3.x and want to use Prefect Cloud Serverless (no workers to manage). What to avoid (Prefect 2.x only): don’t use

Deployment.build_from_flow()

or

prefect deployment build

. In 3.x use

flow.from_source(...).deploy(...)

or

prefect deploy

. 1) Project layout and code organization - Keep flows, tasks, and utilities separate for reuse and testability: - flows/ for orchestration (@flow) only - tasks/ for atomic units (@task) with retries, timeouts, and logging - utils/ for pure helpers (no Prefect imports) - tests/ for unit + integration tests - prefect.yaml in repo root for deployment config (optional but recommended) - Keep your flow entrypoints stable and explicit: - Use file path + function:

flows/etl.py:daily_etl

- Or module path:

package.flows.etl:daily_etl

Example structure:

my_project/
├─ flows/
│  ├─ etl.py
│  └─ monitoring.py
├─ tasks/
│  ├─ io.py
│  └─ transform.py
├─ utils/
├─ tests/
├─ prefect.yaml
└─ pyproject.toml or requirements.txt

2) Flow/task design - Orchestrate in flows; do the work in tasks. - Keep tasks small and idempotent; set retries/timeouts at the task level. - Use

get_run_logger()

for structured logs. Turn on

log_prints=True

in flows if you rely on prints.

from prefect import flow, task
from prefect.logging import get_run_logger

@task(retries=3, retry_delay_seconds=10)
def fetch(url: str) -> dict:
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Fetching {url}")
    ...

@flow(log_prints=True)
def daily_etl(url: str):
    data = fetch(url)
    ...

3) Deployments for Serverless Compute Two great options; pick based on your dependency story: - Source-based deployments (simplest) - Keep code in Git (public or private with creds). - Deploy directly from your repo via

flow.from_source(...).deploy(...)

. - Best when your dependencies are pure-Python or minimal. Example:

from prefect import flow

if __name__ == "__main__":
    flow.from_source(
        source="<https://github.com/org/repo.git>",
        entrypoint="flows/etl.py:daily_etl"
    ).deploy(
        name="daily-etl",
        work_pool_name="my-serverless-pool"
    )

- Image-based deployments (for complex/system deps) - Build a Docker image with all requirements and system libs. - Point your deployment at that image. - Best when you need compiled deps (e.g., database drivers, ML libs). Example (Python API):

from prefect import flow
from prefect.docker import DockerImage

@flow
def daily_etl():
    ...

if __name__ == "__main__":
    daily_etl.deploy(
        name="daily-etl",
        work_pool_name="my-serverless-pool",
        image=DockerImage(
            name="<http://ghcr.io/my-org/my-image:latest|ghcr.io/my-org/my-image:latest>",
            dockerfile="Dockerfile",
            platform="linux/amd64"  # safest default for serverless
        ),
        push=True
    )

You can also define deployments in prefect.yaml and run

prefect deploy

in CI for repeatability:

name: my-project
deployments:
  - name: daily-etl
    entrypoint: flows/etl.py:daily_etl
    work_pool:
      name: my-serverless-pool
    schedule:
      cron: "0 8 * * *"

Then:

prefect deploy

Docs: - How to create deployments - Deploy via Python - Define deployments with prefect.yaml - Store/run code from remote sources 4) Prefect Cloud Serverless specifics - Work pool: Create or use a “Serverless” work pool in Prefect Cloud and target it in your deployment’s

work_pool_name

. - Entrypoints: Must include function, e.g.

flows/etl.py:daily_etl

.

- Images: Prefer

linux/amd64

for compatibility across serverless backends. - Resources/time: Configure memory/CPU via the pool or deployment job variables; keep long-running work chunked (large single runs may hit provider limits). Consider subflows or splitting steps for resiliency. - Networking: Ensure outbound egress is allowed to your data systems; for private resources use VPC/VNet-friendly patterns or APIs. Docs: - Work pools overview - Deployments overview - (Serverless guide) Run code from remote sources and the serverless section under work pools 5) Dependencies - For source-based deployments, prefer pure-Python deps. If you need compiled/system libs, switch to image-based deployments. - For image-based, pin versions and build from a slim base:

FROM python:3.12-slim
RUN apt-get update && apt-get install -y libpq-dev build-essential && rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . /app
WORKDIR /app

- Use a modern Python (3.10+ recommended). 6) Configuration, environment, and secrets - Put non-secret config in environment variables; inject via deployment/job variables. - Store secrets in Prefect Blocks and load at runtime:

from prefect.blocks.system import Secret
db_password = Secret.load("db-password").get()

- Keep credentials out of code and CI logs. Docs: - Store secrets - Manage settings 7) Observability and artifacts - Use

get_run_logger()

everywhere; set

log_prints=True

on flows if needed. - Emit artifacts for key outputs/links (dashboards, S3 locations) to aid debugging. Docs: Create artifacts 8) Testing and CI/CD - Unit test task logic via `.fn`:

def test_transform():
    from tasks.transform import transform_data
    assert transform_data.fn({"x": 2}) == {"x": 4}

- Use the test harness for integration tests:

from prefect.testing.utilities import prefect_test_harness

def test_flow():
    with prefect_test_harness():
        from flows.etl import daily_etl
        daily_etl()

- CI: run tests, then

prefect deploy

to update deployments from main. Docs: Test workflows and CI/CD patterns Quick CLI you’ll use - Validate the command structure (already checked):

prefect deploy --help
prefect deploy                          # all deployments in prefect.yaml
prefect deploy flows/etl.py:daily_etl   # one-off by entrypoint
prefect deploy -n daily-etl             # by name

Common pitfalls to avoid - Don’t use deprecated 2.x build commands or APIs. - Keep entrypoints stable; refactors that move/rename functions without updating deployments will break runs. - For serverless, avoid heavyweight system deps unless you provide an image. - Treat long-running jobs as pipelines of smaller runs/subflows. If you share a bit more about your setup, I can tailor this into a concrete example: - Will your code live in a public or private Git repo? - Do you need system/compiled dependencies (e.g., database drivers, ML)? - Any network/VPC constraints to reach your data systems? - Do you prefer source-based or image-based deployments? References - Deployments (concepts) - Create deployments (how-to) - Deploy via Python - prefect.yaml

- Store/run code from remote sources - Work pools - Test workflows - Artifacts

Love this stack. Here’s a “golden” project layout and patterns that keep things clean, reproducible, and cloud-native with image-based deployments. Top-level goals - Clear separation of concerns: infra (Pulumi), orchestration (Prefect), transformation (dbt) - Zero secrets in repo; blocks and CI secrets only - Single, reproducible Docker image that contains everything needed to run the pipeline - GitHub Actions to build/push image and run

prefect deploy

against Prefect Cloud Serverless Recommended repository structure

elt-project/
├─ orchestration/                 # Prefect flows, tasks, and deployment config
│  ├─ flows/
│  │  ├─ elt.py                  # main pipeline flow (extract -> load -> dbt -> publish)
│  │  └─ utilities.py
│  ├─ tasks/
│  │  ├─ extract.py              # API/file ingestion → GCS
│  │  ├─ load.py                 # GCS → BigQuery load jobs
│  │  └─ artifacts.py            # create Prefect artifacts (links/markdown)
│  ├─ deploy.py                  # optional: Python-based deploy helper (if not using YAML)
│  ├─ prefect.yaml               # build/push steps + deployments (preferred)
│  └─ tests/
│     ├─ test_flows.py
│     └─ test_tasks.py
├─ transformations/               # dbt project
│  ├─ dbt_project.yml
│  ├─ models/
│  │  ├─ staging/
│  │  └─ marts/
│  └─ seeds/
├─ infra/                         # Pulumi program to provision GCP (Bucket + BigQuery)
│  ├─ Pulumi.yaml
│  ├─ Pulumi.dev.yaml            # stack config (use secrets for sensitive values)
│  └─ __main__.py                # or index.ts if you prefer TypeScript
├─ docker/
│  └─ Dockerfile                 # image for Prefect Serverless runs (linux/amd64)
├─ .github/workflows/
│  └─ ci-deploy.yaml             # build/push image + prefect deploy
├─ pyproject.toml or requirements.txt
├─ .env.example                  # non-secret config example (for local dev)
└─ README.md

Key conventions - GCS layout:

gs://<bucket>/{raw|staging|curated}/...

- BigQuery datasets:

raw

,

staging

,

marts

(aligns with dbt models) - Entrypoints stable over time:

orchestration/flows/elt.py:main_elt

- Image-based deployments only (matches your requirement) Dockerfile (image-based deployments)

FROM python:3.12-slim

# System deps for dbt-bigquery/pyarrow, etc.
RUN apt-get update && apt-get install -y \
    git build-essential gcc curl \
  && rm -rf /var/lib/apt/lists/*

# Python deps
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy source
COPY orchestration/ /app/orchestration/
COPY transformations/ /app/transformations/

# Ensure dbt discovers the project in /app/transformations at runtime
ENV DBT_PROJECT_DIR=/app/transformations

# Default to amd64 for broad serverless compatibility
# (also set platform in your build/push if using multi-arch builders)

requirements.txt (example)

prefect>=3.0.0
prefect-docker>=0.4.0
prefect-gcp>=0.5.0
prefect-dbt>=0.5.0
dbt-core>=1.7.0
dbt-bigquery>=1.7.0
google-cloud-bigquery>=3.10.0
google-cloud-storage>=2.13.0
pandas>=2.2.0
pyarrow>=15.0.0

Prefect deployment definition (prefect.yaml) - Uses official Prefect Docker build/push steps and a serverless work pool ``` name: elt-project prefect-version: 3.0.0 build: - prefect_docker.deployments.steps.build_docker_image: image_name: ghcr.io/&lt;org&gt;/elt-project tag: "{{ git-repo.sha }}" dockerfile: docker/Dockerfile # For serverless backends that prefer/require amd64 platform: linux/amd64 push: - prefect_docker.deployments.steps.push_docker_image: image_name: "{{ build-image.image_name }}" tag: "{{ build-image.tag }}" # Optional: set working directory inside the image (if you need it) pull: - prefect.deployments.steps.set_working_directory: directory: /app deployments: - name: daily-elt entrypoint: orchestration/flows/elt.py:main_elt work_pool: name: <your-serverless-pool> # Prefect Cloud Serverless work pool job_variables:

# Inject non-secret env vars here; use Prefect blocks for secrets env: DBT_TARGET: prod DBT_PROJECT_DIR: /app/transformations schedule: cron: "0 7 * * *"

Orchestration: main flow calling extract → load → dbt → publish

# orchestration/flows/elt.py from prefect import flow, task from prefect.logging import get_run_logger from orchestration.tasks.extract import fetch_to_gcs from orchestration.tasks.load import load_gcs_to_bigquery from orchestration.tasks.artifacts import link_looker_dashboard from prefect_dbt.cli import DbtCliProfile, BigQueryTargetConfigs, GlobalConfigs from prefect_dbt.cli.commands import run_dbt_build from prefect_gcp.credentials import GcpCredentials @task def run_dbt(project_dir: str, dataset: str, gcp_creds_block: str, target: str = "prod"): # Load Prefect block with SA creds (users create their own) gcp_creds = GcpCredentials.load(gcp_creds_block) # dbt profile in-memory for BigQuery, no profiles.yml committed profile = DbtCliProfile( name="elt", target=target, target_configs=BigQueryTargetConfigs( project=gcp_creds.project, dataset=dataset, location="US", # Credentials come from the Prefect block’s service_account_info # which dbt-bigquery will pick up via env in this process ), ) # Optionally set global configs (e.g., logs dir) globals = GlobalConfigs() # Execute

dbt build

result = run_dbt_build( project_dir=project_dir, profiles=profile, global_configs=globals ) return result @flow(log_prints=True) def main_elt( source_url: str = "https://example.com/data.csv", gcs_bucket: str = "my-elt-bucket", raw_path: str = "raw/daily/data.csv", bq_dataset_raw: str = "raw", bq_dataset_marts: str = "marts", gcp_creds_block: str = "gcp-credentials", # Prefect block name looker_url: str = "https://lookerstudio.google.com/..." # doc: customize per user ): logger = get_run_logger() # 1) Extract → GCS gcs_uri = fetch_to_gcs(source_url, gcs_bucket, raw_path) # 2) Load GCS → BigQuery (raw) table_id = load_gcs_to_bigquery( gcs_uri=gcs_uri, dataset=bq_dataset_raw, table="daily_ingest", gcp_creds_block=gcp_creds_block ) logger.info(f"Loaded to {table_id}") # 3) Transform with dbt (staging, marts, tests) dbt_result = run_dbt( project_dir="/app/transformations", dataset=bq_dataset_marts, gcp_creds_block=gcp_creds_block, target="prod" ) logger.info(f"dbt build status: {dbt_result}") # 4) Publish: link artifact to Looker Studio dashboard link_looker_dashboard(looker_url) return {"table": table_id, "dbt_status": "completed"}

Example tasks (GCS upload and BQ load)

# orchestration/tasks/extract.py from prefect import task from google.cloud import storage import requests import io @task(retries=3, retry_delay_seconds=10) def fetch_to_gcs(source_url: str, bucket: str, object_name: str) -> str: r = requests.get(source_url, timeout=60) r.raise_for_status() client = storage.Client() # uses GOOGLE_APPLICATION_CREDENTIALS inside the container blob = client.bucket(bucket).blob(object_name) blob.upload_from_file(io.BytesIO(r.content), rewind=True) return f"gs://{bucket}/{object_name}" ``` ``` # orchestration/tasks/load.py from prefect import task from google.cloud import bigquery from google.cloud import storage from prefect_gcp.credentials import GcpCredentials @task(retries=3, retry_delay_seconds=10) def load_gcs_to_bigquery(gcs_uri: str, dataset: str, table: str, gcp_creds_block: str) -> str: creds = GcpCredentials.load(gcp_creds_block).get_credentials_from_service_account() bq = bigquery.Client(credentials=creds) job = bq.load_table_from_uri( source_uris=[gcs_uri], destination=f"{bq.project}.{dataset}.{table}",

job_config=bigquery.LoadJobConfig( write_disposition="WRITE_TRUNCATE", source_format=bigquery.SourceFormat.CSV, autodetect=True, ), ) job.result() return f"{bq.project}.{dataset}.{table}"

Looker link artifact

# orchestration/tasks/artifacts.py from prefect import task from prefect.artifacts import create_link_artifact @task def link_looker_dashboard(url: str): create_link_artifact(key="looker-dashboard", link=url, link_text="Open dashboard")

Pulumi (GCP bucket + BigQuery)
- Keep Pulumi secrets out of the repo; use Pulumi Service backend or a GCS backend with encryption.
- Export outputs you’ll feed into Prefect or dbt as needed (bucket name, project, dataset names).
- Example resources (Python):

# infra/__main__.py import pulumi import pulumi_gcp as gcp project = pulumi.Config("gcp").require("project") location = pulumi.Config().get("location") or "US" bucket = gcp.storage.Bucket( "elt-bucket", location=location, uniform_bucket_level_access=True ) raw = gcp.bigquery.Dataset("raw", dataset_id="raw", location=location) marts = gcp.bigquery.Dataset("marts", dataset_id="marts", location=location) pulumi.export("bucket_name", bucket.name) pulumi.export("project", project) pulumi.export("dataset_raw", raw.dataset_id) pulumi.export("dataset_marts", marts.dataset_id)

dbt project basics (BigQuery)
- Use dbt variables (or env vars) to parameterize datasets.
- Do not commit profiles.yml; let Prefect supply credentials via `DbtCliProfile`.

CI/CD (GitHub Actions)
- Build/push image on main, then `prefect deploy` against your Prefect Cloud account.
- Secrets to set in repo: `PREFECT_API_KEY`, `PREFECT_API_URL`, `GHCR_TOKEN` (or Docker Hub creds).

name: Build and Deploy Prefect on: push: branches: [ main ] jobs: build-and-deploy: permissions: contents: read packages: write id-token: write runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - uses: docker/login-action@v3 with: registry: ghcr.io username: ${{ github.actor }} password: ${{ secrets.GHCR_TOKEN }} - uses: actions/setup-python@v5 with: python-version: "3.12" - name: Install Prefect + Docker extras run: | pip install -r requirements.txt - name: Deploy to Prefect env: PREFECT_API_KEY: ${{ secrets.PREFECT_API_KEY }} PREFECT_API_URL: ${{ secrets.PREFECT_API_URL }} run: | prefect deploy

Secrets and credentials (public repo safe pattern)
- GCP: create a Prefect block holding your service account JSON (not in repo):
  - In code (one-time):

from prefect_gcp.credentials import GcpCredentials import json, os # Read from env or local file; then save as a Prefect block creds = GcpCredentials(service_account_info=json.loads(os.environ["GCP_SA_JSON"])) creds.save("gcp-credentials", overwrite=True)

- Or create via UI. Each user replicating the project creates their own `gcp-credentials` block.
- dbt: no `profiles.yml` in repo. Provide a README step telling users to:
  - create the `gcp-credentials` Prefect block
  - run Pulumi to create bucket/datasets
  - set their Prefect API key/URL in CI or local env
- Prefect deployment: `prefect deploy` reads prefect.yaml and registers a deployment pointing to your serverless pool and the built image.

Local dev ergonomics
- Provide `.env.example` with non-secret defaults (e.g., dataset names).
- Use `pyproject.toml` and pre-commit hooks for linting/formatting.
- Add unit tests for tasks via `.fn` and integration tests with the Prefect test harness:

from prefect.testing.utilities import prefect_test_harness def test_main_elt_smoke(): with prefect_test_harness(): from orchestration.flows.elt import main_elt main_elt.with_options(name="test-run")(source_url="https://example.com/small.csv") ```

Naming and conventions - Flows: snake_case, verbs for tasks (e.g.,

fetch_to_gcs

,

load_gcs_to_bigquery

) - Datasets/tables aligned with dbt layers (raw/staging/marts) - One deployment per flow or environment; keep entrypoints stable How users replicate 1) Fork repo. 2) Provision GCP with Pulumi (

pulumi up

) to create bucket + datasets (using their own project). 3) Create a

gcp-credentials

Prefect block (via UI or code) with their GCP SA JSON. 4) Set Prefect Cloud credentials in CI or local shell. 5) Enable GitHub Packages (or Docker Hub) and set registry creds. 6) Push to main; CI builds image and runs

prefect deploy

. 7) Start the deployment’s schedule or trigger ad hoc runs. Useful docs - Prefect deployments: https://docs.prefect.io/v3/concepts/deployments/ - Deploy via Python: https://docs.prefect.io/v3/how-to-guides/deployments/deploy-via-python/ - prefect.yaml: https://docs.prefect.io/v3/how-to-guides/deployments/prefect-yaml/ - Run flows in Docker: https://docs.prefect.io/v3/how-to-guides/deployment_infra/docker/ - Serverless work pools: https://docs.prefect.io/v3/concepts/work-pools/ - Prefect GCP collection: https://prefecthq.github.io/prefect-gcp/ - Prefect dbt collection: https://prefecthq.github.io/prefect-dbt/ - Pulumi GCP BigQuery: https://www.pulumi.com/registry/packages/gcp/api-docs/bigquery/dataset/ and Storage: https://www.pulumi.com/registry/packages/gcp/api-docs/storage/bucket/ Command you’ll use - We verified this exists and is the right entrypoint:

prefect deploy

If you want, I can turn this into a starter repo (Dockerfile, prefect.yaml, minimal dbt project, Pulumi program, and a working flow) tailored to your org/registry/pool names. Which Pulumi language do you prefer (Python or TypeScript)?