Retrieve code from storage

When a deployment runs, the execution environment needs access to the flow code. Flow code is not stored directly in Prefect server or Prefect Cloud; instead, it must be made available to the execution environment. There are two main ways to achieve this:

1. Include source code directly in your runtime: Often, this means building your code into a Docker image.
2. Retrieve code from storage at runtime: The worker pulls code from a specified location before starting the flow run.

This page focuses on the second approach: retrieving code from a storage location at runtime.

You have several options for where your code can be stored and pulled from:

• Local filesystem
• Git-based storage (GitHub, GitLab, Bitbucket)
• Blob storage (AWS S3, Azure Blob Storage, GCP GCS)

The ideal choice depends on your team’s needs and tools.

In the examples below, we show how to create a deployment configured to run on dynamic infrastructure for each of these storage options.

​

Deployment creation options

As detailed in the Deployment overview, you can create a deployment in one of two main ways:

• Python code with the flow.deploy method

  • When using .deploy, specify a storage location for your flow with the flow.from_source method.

  • The source is either a URL to a git repository or a storage object. For example:

    • A local directory: source=Path(__file__).parent or source="/path/to/file"
    • A URL to a git repository: source="https://github.com/org/my-repo.git"
    • A storage object: source=GitRepository(url="https://github.com/org/my-repo.git")

  • The entrypoint is the path to the file the flow is located in and the function name, separated by a colon.

• YAML specification defined in a prefect.yaml file

  • To create a prefect.yaml file interactively, run prefect deploy from the CLI and follow the prompts.

  • The prefect.yaml file may define a pull section that specifies the storage location for your flow. For example:

    • Set the working directory:

    pull:
        - prefect.deployments.steps.set_working_directory:
            directory: /path/to/directory
    

    • Clone a git repository:

    pull:
        - prefect.deployments.steps.git_clone:
            repository: https://github.com/org/my-repo.git
    

    • Pull from blob storage:

    pull:
        - prefect.deployments.steps.pull_from_blob_storage:
            container: my-container
            folder: my-folder
    

Whether you use from_source or prefect.yaml to specify the storage location for your flow code, the resulting deployment will have a set of pull steps that your worker will use to retrieve the flow code at runtime.

​

Store code locally

If using a Process work pool, you can use one of the remote code storage options shown above, or you can store your flow code in a local folder.

Here is an example of how to create a deployment with flow code stored locally:

from prefect import flow
from pathlib import Path


@flow(log_prints=True)
def my_flow(name: str = "World"):
    print(f"Hello {name}!")


if __name__ == "__main__":
    my_flow.from_source(
        source=str(Path(__file__).parent),  # code stored in local directory
        entrypoint="local_process_deploy_local_code.py:my_flow",
    ).deploy(
        name="local-process-deploy-local-code",
        work_pool_name="my-process-pool",
    )

​

Git-based storage

Git-based version control platforms provide redundancy, version control, and collaboration capabilities. Prefect supports:

• GitHub
• GitLab
• Bitbucket

For a public repository, you can use the repository URL directly.

If you are using a private repository and are authenticated in your environment at deployment creation and deployment execution, you can use the repository URL directly.

Alternatively, for a private repository, you can create a Secret block or git-platform-specific credentials block to store your credentials:

• GitHubCredentials
• BitBucketCredentials
• GitLabCredentials

Then you can reference this block in the Python deploy method or the prefect.yaml file pull step.

If using the Python deploy method with a private repository that references a block, provide a GitRepository object instead of a URL, as shown below.

from prefect import flow

if __name__ == "__main__":
    flow.from_source(
        source="https://github.com/org/my-public-repo.git",
        entrypoint="gh_public_repo.py:my_flow",
    ).deploy(
        name="my-github-deployment",
        work_pool_name="my_pool",
    )

from prefect import flow

if __name__ == "__main__":
    flow.from_source(
        source="https://github.com/org/my-public-repo.git",
        entrypoint="gh_public_repo.py:my_flow",
    ).deploy(
        name="my-github-deployment",
        work_pool_name="my_pool",
    )

from prefect import flow


if __name__ == "__main__":
    flow.from_source(
        source="https://bitbucket.com/org/my-public-repo.git",
        entrypoint="bb_public_repo.py:my_flow",
    ).deploy(
        name="my-bitbucket-deployment",
        work_pool_name="my_pool",
    )

from prefect import flow


if __name__ == "__main__":
    gitlab_repo = "https://gitlab.com/org/my-public-repo.git"

    flow.from_source(
        source=gitlab_repo,
        entrypoint="gl_public_repo.py:my_flow"
    ).deploy(
        name="my-gitlab-deployment",
        work_pool_name="my_pool",
    )

Note that you can specify a branch if creating a GitRepository object. The default is "main".

​

Docker-based storage

Another popular flow code storage option is to include it in a Docker image. All work pool options except Process and Prefect Managed allow you to bake your code into a Docker image.

To create a deployment with Docker-based flow code storage use the Python deploy method or create a prefect.yaml file.

The prefect.yaml file below was generated by running prefect deploy from the CLI (a few lines of metadata were excluded from the top of the file output for brevity).

Note that the build section is necessary if baking your flow code into a Docker image.

from prefect import flow


@flow
def my_flow():
    print("Hello from inside a Docker container!")

if __name__ == "__main__":
    my_flow.deploy(
        name="my-docker-deploy",
        work_pool_name="my_pool",
        image="my-docker-image:latest",
        push=False
    )

By default, .deploy will build a Docker image that includes your flow code and any pip packages specified in a requirements.txt file.

In the examples above, we elected not to push the resulting image to a remote registry.

To push the image to a remote registry, pass push=True in the Python deploy method or add a push_docker_image step to the push section of the prefect.yaml file.

​

Custom Docker image

If an image is not specified by one of the methods above, deployment flow runs associated with a Docker work pool will use the base Prefect image (e.g. prefecthq/prefect:3-latest) when executing. Alternatively, you can create a custom Docker image outside of Prefect by running docker build && docker push elsewhere (e.g. in your CI/CD pipeline) and then reference the resulting image in the job_variables section of your deployment definition, or set the image as a default directly on the work pool.

For more information, see this discussion of custom Docker images.

​

Blob storage

Another option for flow code storage is any fsspec-supported storage location, such as AWS S3, Azure Blob Storage, or GCP GCS.

If the storage location is publicly available, or if you are authenticated in the environment where you are creating and running your deployment, you can reference the storage location directly. You don’t need to pass credentials explicitly.

To pass credentials explicitly to authenticate to your storage location, you can use either of the following block types:

• Prefect integration library storage blocks, such as the prefect-aws library’s S3Bucket block, which can use a AWSCredentials block when it is created.
• Secret blocks

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

The examples below show how to create a deployment with flow code in a cloud provider storage location. For each example, we show how to access code that is publicly available. The prefect.yaml example includes an additional line to reference a credentials block if authenticating to a private storage location through that option.

We also include Python code that shows how to use an existing storage block and an example of that creates, but doesn’t save, a storage block that references an existing nested credentials block.

from prefect import flow


if __name__ == "__main__":
    flow.from_source(
        source="s3://my-bucket/my-folder",
        entrypoint="my_file.py:my_flow",
    ).deploy(
        name="my-aws-s3-deployment",
        work_pool_name="my-work-pool"
    )

from prefect import flow


if __name__ == "__main__":
    flow.from_source(
        source="s3://my-bucket/my-folder",
        entrypoint="my_file.py:my_flow",
    ).deploy(
        name="my-aws-s3-deployment",
        work_pool_name="my-work-pool"
    )

from prefect import flow


if __name__ == "__main__":
    flow.from_source(
        source="az://mycontainer/myfolder",
        entrypoint="my_file.py:my_flow",
    ).deploy(
        name="my-azure-deployment",
        work_pool_name="my-work-pool",
        job_variables={"env": {"EXTRA_PIP_PACKAGES": "prefect-azure"} }, 
    )

from prefect import flow


if __name__ == "__main__":
    flow.from_source(
        source="gs://my-bucket/my-folder",  
        entrypoint="my_file.py:my_flow",
    ).deploy(
        name="my-gcs-deployment",
        work_pool_name="my-work-pool"
    )

Another authentication option is to give the worker access to the storage location at runtime through SSH keys.