✅ thought for 299.7 seconds

Short answer: set the branch on the storage you pass to

flow.from_source(...)

. In Prefect 3.x,

Flow.from_source

only accepts

source

and

entrypoint

. To choose a git branch, pass a

GitRepository

with

branch=...

(or pin to a

commit_sha

). Example:

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

# entrypoint is "path/to/file.py:flow_function_name"
my_flow = flow.from_source(
    source=GitRepository(
        url="<https://github.com/your-org/your-repo.git>",
        branch="feature/sparkle",              # or a tag name
        # commit_sha="a1b2c3d4...",           # takes precedence over branch if set
        # include_submodules=True,
        # credentials=...,                    # token/user if private
    ),
    entrypoint="flows/daily_job.py:daily_job",
)

- GitHub or Bitbucket: use the normal git HTTPS URL;

branch

works the same for both. - Entrypoint must be

relative/path/to/file.py:flow_name

within the repo. - If you need to pin an exact version, use

commit_sha

instead of

branch

. Deploying it:

await my_flow.deploy(
  name="daily-job-prod",
  work_pool_name="my-pool"
)

Notes and references - The relevant bits live in

prefect.runner.storage.GitRepository

(it has

branch

and

commit_sha

) and the git clone step that consumes them: - storage.py - pull.py (

git_clone

supports

branch

and

commit_sha

) - In 3.x there is no

Deployment.build_from_flow()

anymore; you use

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

. If you were hoping to cram the branch into a magic URL string, don’t. The explicit

GitRepository(..., branch="...")

is the supported, unsurprising way.