How to retrieve code from storage

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.

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

​

Deployment creation options

• 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 either the path to the file and the function name separated by a colon (for example, my_flow.py:my_func), or a Python module path (for example, my_module.my_flow.my_func).
• 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
    

​

Store code 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

• GitHub
• GitLab
• Bitbucket

• GitHubCredentials
• BitBucketCredentials
• GitLabCredentials

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",
    )

​

Prefect Cloud GitHub App integration

​

How it works

1. An organization owner installs the app and grants it access to selected repositories.
2. At deployment runtime, the worker calls prefect-cloud github token to request a short-lived installation access token from Prefect Cloud.
3. Prefect Cloud exchanges its credentials with GitHub to generate the token, scoped only to the repositories the app can access.
4. The worker uses this token to clone the repository, then the token expires.

​

Security details for InfoSec review

​

Set up the GitHub App

pull:
    - prefect.deployments.steps.run_shell_script:
        id: get-github-token
        script: uv tool run prefect-cloud github token owner/repository
        stream_output: false
    - prefect.deployments.steps.git_clone:
        id: git-clone
        repository: https://x-access-token:{{ get-github-token.stdout }}@github.com/owner/repository.git

​

Docker-based storage

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
    )

​

Custom Docker image

​

Blob storage

• 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

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"
    )