Short answer: not built in for git-based source pulls in Prefect 3.x. If your deployment uses

flow.from_source("<https://gitlab>...")

, Prefect will try to fetch/update that repo at run time. If GitLab is unreachable, the pull step can fail the run. There isn’t a native “ignore git errors and use whatever is already on disk” switch. A few useful details: - Prefect does retry git operations a few times - but if it still can’t reach the remote, it does not automatically fall back to a cached checkout - so there’s no supported “skip git pull if GitLab is down” option for standard git-sourced deployments Recommended alternatives 1. Bake code into the image Best option if you want runs to be independent of GitLab availability. - Build a Docker image containing your flow code - Deploy using that image - Then workers don’t need to pull code from git at run time 2. Use a custom

pull

step with fallback logic You can define your own step in

prefect.yaml

that: - tries

git pull

- if that fails, uses an existing local checkout - errors only if no cached copy exists 3. Use non-git remote storage For example S3 / Azure / GCS as the source, if that’s more reliable in your environment. 4. Use local code on the worker If the worker always has the repo mounted locally, you can avoid runtime git fetches entirely, but then you are responsible for keeping the code updated. If you want, I can show you either: - a custom

prefect.yaml

pull step that falls back to a local checkout, or - the recommended Docker-image deployment pattern for Prefect 3.x. Related docs: - Store flow code for deployments - Deploy via Python - Deploy with Docker

Yep — here’s a working pattern for a custom pull step with fallback to a local checkout in Prefect 3.x. How it works - Prefect runs your custom pull step on the worker - the step tries to clone/update from GitLab - if GitLab is unavailable, it uses an existing local directory - the step returns

{"directory": "..."}

so Prefect knows where to run the flow from 1) Create a custom step module For example, save this as

my_steps.py

somewhere available to the worker process:

from pathlib import Path
import subprocess
from typing import Any, Optional

from prefect.logging.loggers import get_logger

logger = get_logger("deployment")


def git_clone_with_fallback(
    repository: str,
    branch: Optional[str] = None,
    fallback_path: Optional[str] = None,
) -> dict[str, Any]:
    """
    Try to clone/update a git repo. If that fails, use a local fallback copy.

    Returns a dict with a "directory" key so Prefect will use that as the
    working directory for the flow run.
    """
    cwd = Path.cwd()

    repo_name = repository.rstrip("/").split("/")[-1]
    if repo_name.endswith(".git"):
        repo_name = repo_name[:-4]

    fresh_clone_path = cwd / repo_name

    try:
        <http://logger.info|logger.info>(f"Trying to clone {repository}")

        cmd = ["git", "clone"]
        if branch:
            cmd.extend(["-b", branch])
        cmd.extend([repository, str(fresh_clone_path)])

        subprocess.run(
            cmd,
            check=True,
            timeout=120,
            capture_output=True,
            text=True,
        )

        <http://logger.info|logger.info>(f"Cloned repo to {fresh_clone_path}")

        return {
            "directory": str(fresh_clone_path),
            "source": "remote",
        }

    except Exception as exc:
        logger.warning(f"Git clone failed: {exc}")

        if not fallback_path:
            raise RuntimeError(
                "Git clone failed and no fallback_path was provided."
            ) from exc

        fallback = Path(fallback_path).expanduser().resolve()

        if not fallback.exists() or not fallback.is_dir():
            raise RuntimeError(
                f"Git clone failed and fallback path does not exist: {fallback}"
            ) from exc

        logger.warning(f"Using fallback repo at {fallback}")

        return {
            "directory": str(fallback),
            "source": "fallback",
        }

2) Reference it in

prefect.yaml

Example `prefect.yaml`:

pull:
  - my_steps.git_clone_with_fallback:
      id: get-code
      repository: <https://gitlab.com/my-group/my-repo.git>
      branch: main
      fallback_path: /opt/prefect/repo-cache/my-repo

  - prefect.deployments.steps.pip_install_requirements:
      directory: "{{ get-code.directory }}"
      requirements_file: requirements.txt

What this does: - first step gets code from GitLab if possible, otherwise

/opt/prefect/repo-cache/my-repo

- second step installs dependencies from the selected directory 3) Your flow entrypoint should point into that repo For example, if your deployment uses:

deployments:
  - name: my-deployment
    entrypoint: flows/my_flow.py:my_flow

then both: - the cloned repo, and - the fallback repo must contain

flows/my_flow.py

. 4) Important caveats - The fallback directory must already exist on the worker - Prefect will not create or maintain that cache for you - Your custom module must be importable by the worker - e.g.

my_steps.py

must be in the working directory or on

PYTHONPATH

- This is a custom workaround, not built-in Prefect git cache behavior - if you need repeatable production behavior, baking code into the image is usually safer 5) Slightly better version: try updating existing fallback repo first If you want, you can make it smarter: - if fallback repo exists, run

git fetch

/

git pull

- if update fails, continue with the existing checkout as-is That usually works better than always doing a fresh clone first. Example: ```

from pathlib import Path import subprocess from typing import Any, Optional from prefect.logging.loggers import get_logger logger = get_logger("deployment") def git_update_with_fallback( repository: str, branch: str = "main", fallback_path: str = "/opt/prefect/repo-cache/my-repo", ) -> dict[str, Any]: repo_dir = Path(fallback_path).expanduser().resolve() if repo_dir.exists() and (repo_dir / ".git").exists(): try: logger.info(f"Trying to update cached repo in {repo_dir}") subprocess.run( ["git", "-C", str(repo_dir), "fetch", "origin"], check=True, timeout=60, capture_output=True, text=True, ) subprocess.run( ["git", "-C", str(repo_dir), "checkout", branch], check=True, timeout=30, capture_output=True, text=True, ) subprocess.run( ["git", "-C", str(repo_dir), "pull", "origin", branch], check=True, timeout=60, capture_output=True, text=True, ) return { "directory": str(repo_dir), "source": "updated-fallback", } except Exception as exc: logger.warning(f"Could not update cached repo: {exc}") logger.warning("Using cached repo as-is") return { "directory": str(repo_dir), "source": "stale-fallback", } try: logger.info(f"Cached repo not found, cloning {repository}") parent = repo_dir.parent parent.mkdir(parents=True, exist_ok=True) subprocess.run( ["git", "clone", "-b", branch, repository, str(repo_dir)], check=True, timeout=120, capture_output=True, text=True, ) return { "directory": str(repo_dir), "source": "fresh-clone", } except Exception as exc: raise RuntimeError( f"Could not clone repository and no usable cached repo exists at {repo_dir}" ) from exc

Then in `prefect.yaml`:

pull: - my_steps.git_update_with_fallback: id: get-code repository: https://gitlab.com/my-group/my-repo.git branch: main fallback_path: /opt/prefect/repo-cache/my-repo - prefect.deployments.steps.pip_install_requirements: directory: "{{ get-code.directory }}" requirements_file: requirements.txt ``` Docs - Customize deployments with `prefect.yaml` - Store flow code for deployments If you want, I can also give you: - a full project layout example (

prefect.yaml

+ flow file + custom step file), or - a version tailored for Docker workers or Kubernetes workers.