Short answer: your suspicion is right. In Prefect 3.x, if you run that script from your repo root with push=True (the default), each deployment will upload/sync the entire current working directory to its GCS prefix. So both gs://my-bucket/my-first-flow and gs://my-bucket/my-second-flow would receive the same full repo unless you limit what gets pushed. Ways to make each deployment only contain its own subfolder: 1) Change the working directory before each deploy This makes the “project root” for the push just that subfolder.

import os
from prefect import flow

def deploy_subdir(subdir, bucket_prefix, entrypoint, name):
    cwd = os.getcwd()
    try:
        os.chdir(subdir)
        flow.from_source(
            source=f"<gs://my-bucket/{bucket_prefix}>",
            entrypoint=entrypoint,  # now relative to `subdir`
        ).deploy(name=name)
    finally:
        os.chdir(cwd)

if __name__ == "__main__":
    deploy_subdir(
        subdir="pipelines/my_first_flow",
        bucket_prefix="my-first-flow",
        entrypoint="my-first-flow.py:my_flow",
        name="my-first-gcs-deployment",
    )

    deploy_subdir(
        subdir="pipelines/my_second_flow",
        bucket_prefix="my-second-flow",
        entrypoint="my-second-flow.py:my_flow",
        name="my-second-gcs-deployment",
    )

Notes: - Your second entrypoint looked like it was missing “.py”. It should be

.../my-second-flow.py:my_flow

. - Entry points are resolved relative to the source root (your cwd when deploying in this pattern). 2) Don’t push from Prefect; pre-stage code yourself If you want deterministic contents with no filtering logic, upload each pipeline folder to its own GCS prefix yourself (e.g., `gsutil -m rsync -r pipelines/my_first_flow gs://my-bucket/my-first-flow`), then deploy with push disabled so Prefect won’t upload anything:

from prefect import flow

if __name__ == "__main__":
    flow.from_source(
        source="<gs://my-bucket/my-first-flow>",
        entrypoint="my-first-flow.py:my_flow",
    ).deploy(name="my-first-gcs-deployment", push=False)

    flow.from_source(
        source="<gs://my-bucket/my-second-flow>",
        entrypoint="my-second-flow.py:my_flow",
    ).deploy(name="my-second-gcs-deployment", push=False)

3) Use a .prefectignore to filter files (with caveats) A

.prefectignore

at your project root can exclude files/folders during the push. However, it’s global for the project—so it’s hard to include only “my_first_flow” for one deployment and only “my_second_flow” for the other in the same run. This is typically easier if you: - Run deployments from each subdirectory (option 1), or - Maintain separate projects/repos, or - Run two deploy commands with different working directories. Additional notes - GCS sources are supported via fsspec/gcsfs. Make sure

gcsfs

is installed and your credentials are available to the deploy and the worker. - Best practice for multi-flow repos with remote storage is to either deploy from each subdirectory (so each push only includes that folder) or to pre-stage code and use

push=False

. Docs and API references - How to store flow code: Store flow code - Runner storage API (Git/Remote/Local): prefect.runner.storage - Deployment pull steps (how code is fetched at runtime): prefect.deployments.steps.pull - Project configuration and push/pull steps: prefect.yaml guide - Filesystem utilities (ignore patterns): prefect.utilities.filesystem Would you prefer the change-directory approach or the push=False + pre-stage approach? I can tailor an example to your exact repo layout.

Thanks for the detailed plan — this is all doable in Prefect 3.x. The main tension is: - You want a single parent-level deployment script so shared utilities can be imported (nice for CI/CD and consistency) - You don’t want every dev deployment to sync the entire repo to GCS You can have both by staging just the needed code for each dev deployment (flow subdir + utilities) before calling deploy, while still keeping a single “all_deployments.py” at the repo root. For prod, you can point deployments to GitHub and only pull the required subdirectories using sparse checkout. Key points to match your design - One Prefect server instance: fine. - Work pools: create workpool-a-dev and workpool-a-prod. Set pool-level env var

pool_env

accordingly in each pool. In your flows, read

os.getenv("pool_env", "DEV")

to switch configs. - Single deployment script: parameterize by

--env

(DEV/PROD) and optional

--flow

to deploy one or all. - DEV: - Suffix work pool with “-dev” - Schedule disabled - Deployment name prefixed with DEV- - Storage: GCS prefix per flow - Only sync the flow’s folder plus your utilities using a temporary staging directory - PROD: - Suffix work pool with “-prod” - Keep schedule - Deployment name prefixed with PROD- - Storage: GitHub main branch (via GitRepository) with sparse directories so only flow folder + utilities are pulled at runtime Working example: all_deployments.py - Assumes your repo structure: - utilities/ (shared code, e.g., common_functions.py) - pipelines/my_first_flow/my-first-flow.py:my_flow - pipelines/my_second_flow/my-second-flow.py:my_flow ``` import argparse import os import shutil import tempfile from pathlib import Path from prefect import flow from prefect.runner.storage import GitRepository # Configure your flows here FLOWS = { "my_first_flow": { "entrypoint": "pipelines/my_first_flow/my-first-flow.py:my_flow", "subdir": "pipelines/my_first_flow", "base_name": "snowflake-to-bigquery-sync", "workpool_base": "workpool-a", # Only used in prod example; for dev we default to schedule disabled "cron": "0 2 * * *", # example }, "my_second_flow": { "entrypoint": "pipelines/my_second_flow/my-second-flow.py:my_flow", "subdir": "pipelines/my_second_flow", "base_name": "another-pipeline", "workpool_base": "workpool-a", "cron": "0 6 * * *", }, } # Put any shared directories you want included with each flow in dev pushes SHARED_DIRS = ["utilities"] # adjust to your repo def stage_dev_build(subdir: str, shared_dirs: list[str]) -> str: """ Create a temp directory that contains only: - the flow's subdirectory - the shared utilities directories Returns the path to the temp directory. Caller is responsible for chdir/cleanup context. """ tmpdir = tempfile.mkdtemp(prefix="prefect-dev-stage-") repo_root = Path.cwd() # Copy flow subdir preserving relative structure src_flow_dir = repo_root / subdir dst_flow_dir = Path(tmpdir) / subdir dst_flow_dir.parent.mkdir(parents=True, exist_ok=True) shutil.copytree(src_flow_dir, dst_flow_dir, dirs_exist_ok=True) # Copy shared dirs to the root of the staging area for d in shared_dirs: src = repo_root / d if src.exists(): dst = Path(tmpdir) / d shutil.copytree(src, dst, dirs_exist_ok=True) return tmpdir def deploy_dev(flow_key: str, cfg: dict, gcs_bucket_prefix: str): """ Deploy a single flow to GCS for DEV. Only copies the flow folder + utilities to staging. """ tmpdir = stage_dev_build(cfg["subdir"], SHARED_DIRS) try: os.chdir(tmpdir) # Each flow goes to its own GCS prefix so they stay isolated gcs_source = f"{gcs_bucket_prefix}/{flow_key}" dep_id = flow.from_source( source=gcs_source, # requires gcsfs to be installed and credentials available

entrypoint=cfg["entrypoint"], # relative to tmpdir we created ).deploy( name=f"DEV-{cfg['base_name']}", work_pool_name=f"{cfg['workpool_base']}-dev", schedule=None, # no schedule for dev push=True, # push staged contents to GCS ) print(f"Deployed DEV {flow_key}: {dep_id}") finally: # Return to repo root os.chdir(str(Path(file).parent)) def deploy_prod(flow_key: str, cfg: dict, git_url: str, branch: str = "main"): """ Deploy a single flow using Git source for PROD. We sparse-checkout only the flow subdir and shared utilities so the worker only pulls the minimum required code. """ storage = GitRepository( url=git_url, branch=branch, directories=[cfg["subdir"], *SHARED_DIRS], # sparse checkout ) dep_id = flow.from_source( source=storage, entrypoint=cfg["entrypoint"], ).deploy( name=f"PROD-{cfg['base_name']}", work_pool_name=f"{cfg['workpool_base']}-prod", # keep schedule for prod cron=cfg.get("cron"), # or use interval/rrule as needed push=False, # nothing to push; workers will git clone at runtime ) print(f"Deployed PROD {flow_key}: {dep_id}") def main(): parser = argparse.ArgumentParser() parser.add_argument("--env", dest="master_env", default="DEV", choices=["DEV", "PROD"]) parser.add_argument("--flow", dest="flow_name", default=None, help="Deploy only this flow key") # Dev GCS and Prod Git config parser.add_argument("--dev-gcs", dest="dev_gcs_prefix", default="gs://my-bucket") parser.add_argument("--prod-git", dest="prod_git_url", default="https://github.com/org/repo") parser.add_argument("--prod-branch", dest="prod_branch", default="main") args = parser.parse_args() # Select which flows to deploy targets = [args.flow_name] if args.flow_name else list(FLOWS.keys()) for key in targets: cfg = FLOWS[key] if args.master_env == "DEV": deploy_dev(key, cfg, gcs_bucket_prefix=args.dev_gcs_prefix) else: deploy_prod(key, cfg, git_url=args.prod_git_url, branch=args.prod_branch) if name == "__main__": main()

How flows read pool_env and configs
- Set a pool-level environment variable on each pool: pool_env=DEV on workpool-a-dev/workpool-b-dev and pool_env=PROD on prod pools.
- In your flows, default to DEV for local runs:

import os from prefect import flow @flow def my_flow(): pool_env = os.getenv("pool_env", "DEV") if pool_env == "PROD": # load prod config ... else: # load dev config ... ``` Why this solves your concerns - Only the flow’s subdir and utilities are uploaded to GCS in dev (not the entire repo), but you still keep a single parent deployment script. - In prod, code is pulled from GitHub. With GitRepository(directories=[...]), workers only pull the flow subdir + utilities, keeping the runtime footprint small and consistent with dev layout. - You can deploy all or a single flow with flags, and you can toggle schedules, names, and work pools based on the

--env

setting. Notes and caveats - GCS usage: Install gcsfs where you run deploys and on workers that will pull from GCS. Ensure GCP credentials are available to both deploy-time and run-time environments. - If you prefer to avoid staging, you could use a .prefectignore file, but it’s global. Staging per-flow is more deterministic for your “one script” goal. - If your utilities are large or reused across many repos, consider packaging them as a Python package and installing via requirements, then your staged set becomes even smaller. References - Store flow code: Store flow code

- Runner storage API (Git/Remote/Local): prefect.runner.storage - Deployment pull steps (how code is fetched at runtime): prefect.deployments.steps.pull - Project config and push/pull overview: prefect.yaml guide - Filesystem utilities (ignore patterns): prefect.utilities.filesystem If you share your exact repo structure (utilities folder name, number of flows), I can tailor the staging step and entrypoint paths precisely.

Yes — using a Git-only workflow with a testing branch is a great way to simplify this. You can point both DEV and PROD deployments at Git, avoid GCS entirely, and still keep a single parent-level deployment script plus shared utilities. How it would look - DEV: - Branch: testing - Work pool: suffixed with -dev - Name prefix: DEV- - Schedule: None - Storage: GitRepository with sparse directories (flow folder + utilities) - PROD: - Branch: main (or pin to a commit_sha for reproducibility) - Work pool: suffixed with -prod - Name prefix: PROD- - Schedule: kept - Storage: GitRepository with the same sparse directories Key benefits - No push/sync to GCS; workers just git clone at runtime. - Minimal code checkout via Git sparse-checkout (

directories=[...]

) so each deployment only pulls its flow subdir + utilities. - Single all_deployments.py at repo root remains intact and simple. Example all_deployments.py

import argparse
from prefect import flow
from prefect.runner.storage import GitRepository

FLOWS = {
    "my_first_flow": {
        "entrypoint": "pipelines/my_first_flow/my-first-flow.py:my_flow",
        "subdir": "pipelines/my_first_flow",
        "base_name": "snowflake-to-bigquery-sync",
        "workpool_base": "workpool-a",
        "cron": "0 2 * * *",  # example
    },
    "my_second_flow": {
        "entrypoint": "pipelines/my_second_flow/my-second-flow.py:my_flow",
        "subdir": "pipelines/my_second_flow",
        "base_name": "another-pipeline",
        "workpool_base": "workpool-a",
        "cron": "0 6 * * *",
    },
}

SHARED_DIRS = ["utilities"]  # common code to include for both envs


def deploy_for_env(env: str, flow_key: str, cfg: dict, git_url: str, branch: str, commit_sha: str | None = None):
    storage = GitRepository(
        url=git_url,
        branch=None if commit_sha else branch,
        commit_sha=commit_sha,
        directories=[cfg["subdir"], *SHARED_DIRS],  # sparse checkout: just flow dir + utilities
    )

    name_prefix = f"{env}-"
    work_pool_name = f"{cfg['workpool_base']}-{env.lower()}"

    dep_id = flow.from_source(
        source=storage,
        entrypoint=cfg["entrypoint"],
    ).deploy(
        name=name_prefix + cfg["base_name"],
        work_pool_name=work_pool_name,
        schedule=None if env == "DEV" else cfg.get("cron"),
        push=False,  # nothing to push; workers will git clone
    )
    print(f"Deployed {env} {flow_key}: {dep_id}")


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--env", default="DEV", choices=["DEV", "PROD"])
    parser.add_argument("--flow", default=None, help="Deploy only this flow key")
    parser.add_argument("--git-url", default="<https://github.com/org/repo>")
    parser.add_argument("--dev-branch", default="testing")
    parser.add_argument("--prod-branch", default="main")
    parser.add_argument("--prod-commit", default=None, help="Optional commit SHA to pin PROD")
    args = parser.parse_args()

    targets = [args.flow] if args.flow else list(FLOWS.keys())

    for key in targets:
        cfg = FLOWS[key]
        if args.env == "DEV":
            deploy_for_env("DEV", key, cfg, args.git_url, branch=args.dev_branch)
        else:
            deploy_for_env("PROD", key, cfg, args.git_url, branch=args.prod_branch, commit_sha=args.prod_commit)


if __name__ == "__main__":
    main()

Notes - Per-flow isolation: directories must include the entrypoint path and any imports (e.g., utilities). Workers only clone those folders. - Credentials: Ensure workers can access the repo (SSH or token). If private, you can pass credentials via GitRepository’s credentials parameter or a block. - Reproducibility: For PROD, consider pinning to a specific commit_sha for deterministic runs. - Your pool_env strategy still works the same: set pool-level env var pool_env=DEV on -dev pools and pool_env=PROD on -prod pools; flows read

os.getenv("pool_env", "DEV")

. Docs

- Store flow code (Git): Store flow code - Runner storage API (GitRepository, directories): prefect.runner.storage - Pull steps (git_clone at runtime): prefect.deployments.steps.pull If you want, share your repo layout and I’ll tailor the sparse-checkout directories and entrypoints.