Great question — and agreed, a fallback mechanism would be handy. Short answer: In Prefect 3.x there isn’t a built-in “try these sources in order” feature for

flow.from_source(...)

or deployment pull steps.

Flow.from_source

accepts a single source, and deployment

pull

steps run sequentially without conditional/fallback semantics — if a step fails, the run fails; there’s no “only run the next step if the previous one failed” behavior. Workable approaches today: - Custom RunnerStorage (recommended): Implement a small storage wrapper that tries GitHub/GitLab first and falls back to a local path. You can pass this to

flow.from_source(...)

, and implement

to_pull_step()

so deployments inherit the same behavior. Example sketch:

from pathlib import Path
from typing import Optional, Any
from prefect.runner.storage import RunnerStorage, GitRepository, LocalStorage
import asyncio

class FallbackStorage(RunnerStorage):
    def __init__(self, sources: list[RunnerStorage]):
        self.sources = sources
        self._chosen: Optional[RunnerStorage] = None
        self._base: Optional[Path] = None

    def set_base_path(self, path: Path) -> None:
        self._base = path
        for s in self.sources:
            s.set_base_path(path)

    @property
    def destination(self) -> Path:
        return self._chosen.destination if self._chosen else (self._base or Path.cwd())

    @property
    def pull_interval(self) -> Optional[int]:
        return self._chosen.pull_interval if self._chosen else None

    async def pull_code(self) -> None:
        last_err = None
        for s in self.sources:
            try:
                await s.pull_code()
                self._chosen = s
                return
            except Exception as e:
                last_err = e
        raise RuntimeError(f"All fallback sources failed; last error: {last_err}")

    def to_pull_step(self) -> dict[str, Any] | list[dict[str, Any]]:
        # If a source has been selected (e.g., at runtime), defer to it
        if self._chosen:
            return self._chosen.to_pull_step()
        # Otherwise, return a composite of all underlying steps — your implementation
        # can wrap these to try them one-by-one and ignore failures until success.
        return [s.to_pull_step() for s in self.sources]

# Usage
from prefect import flow

storage = FallbackStorage([
    GitRepository(url="<https://gitlab.com/org/repo.git>", branch="main"),
    GitRepository(url="<https://github.com/org/repo.git>", branch="main"),
    LocalStorage(path="/opt/prefect/code-fallback"),
])

f = flow.from_source(source=storage, entrypoint="flows.py:my_flow")
f.deploy(name="my-deployment", work_pool_name="your-pool")

Notes: - This gives you true fallback at runtime. You may want to enhance

to_pull_step()

to emit a single custom step that encapsulates the fallback try-order, so that deployments behave the same way when workers execute pull steps. - Make sure your “local” fallback path actually exists on the worker/container image. If you prefer, I can help turn this into a small reusable utility, or we can open a feature request to add native fallback semantics to sources/pull steps. Would you like me to start a GitHub discussion/issue on your behalf with this proposal (GitLab/GitHub first, then local path), and include your use case? If so, are you primarily deploying with

prefect deploy

and workers, or constructing deployments programmatically with

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

?

Sorry, I encountered an error while processing your request. Please try again.

have you tried writing a custom step? / is there a reason that doesn't work for you?