prefect.yaml

.deploy()

prefect.yaml

.github/workflows/

PREFECT_API_KEY

PREFECT_API_URL

DOCKER_USERNAME

DOCKER_PASSWORD

.deploy()

yaml
name: Deploy Prefect flow
on:
  push:
    branches:
      - main

jobs:
  deploy:
    name: Deploy
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v4

      - name: Log into Docker Hub
        uses: docker/login-action@v2
        with:
          username: ${{ secrets.DOCKER_USERNAME }}
          password: ${{ secrets.DOCKER_PASSWORD }}

      - name: Set up Python
        uses: actions/setup-python@v2
        with:
          python-version: '3.x'
          
      - name: Install Dependencies
        run: |
          pip install -r requirements.txt

      - name: Deploy Flow
        env:
          PREFECT_API_KEY: ${{ secrets.PREFECT_API_KEY }}
          PREFECT_API_URL: ${{ secrets.PREFECT_API_URL }}
        run: |
          python -c "\
          from your_script import your_flow_function ;\
          your_flow_function().deploy()"

.deploy()

deploy_flows.py

python
import os
import importlib

flow_directory = 'path/to/your/flows'

for filename in os.listdir(flow_directory):
    if filename.endswith('.py'):
        module_name = filename[:-3]
        module_path = f"{flow_directory.replace('/', '.')}.{module_name}"
        
        module = importlib.import_module(module_path)
        
        # Assuming each flow script has a function `get_flow()`
        flow = getattr(module, 'get_flow')()
        flow.deploy()

yaml
name: Deploy Prefect flows
on:
  push:
    branches:
      - main

jobs:
  deploy:
    name: Deploy
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v4

      - name: Log into Docker Hub
        uses: docker/login-action@v2
        with:
          username: ${{ secrets.DOCKER_USERNAME }}
          password: ${{ secrets.DOCKER_PASSWORD }}

      - name: Set up Python
        uses: actions/setup-python@v2
        with:
          python-version: '3.x'
          
      - name: Install Dependencies
        run: |
          pip install -r requirements.txt

      - name: Deploy All Flows
        env:
          PREFECT_API_KEY: ${{ secrets.PREFECT_API_KEY }}
          PREFECT_API_URL: ${{ secrets.PREFECT_API_URL }}
        run: |
          python path/to/deploy_flows.py

.deploy()

python
from datetime import timedelta, datetime
from prefect.client.schemas.schedules import IntervalSchedule

your_flow.deploy(
    name="your_flow_deployment_name",
    schedule=IntervalSchedule(
        interval=timedelta(minutes=10),
        anchor_date=datetime(2026, 1, 1, 0, 0),
        timezone="America/Chicago"
    )
)

python
from prefect.client.schemas.schedules import CronSchedule

your_flow.deploy(
    name="your_flow_deployment_name",
    schedule=CronSchedule(
        cron="0 0 * * *",
        timezone="America/Chicago"
    )
)

Variable.get()

python
from prefect import flow
from prefect.variables import Variable

@flow
def my_sync_flow():
    # Set a variable
    Variable.set(name="my_variable", value="Hello, Prefect!")
    
    # Get the variable value
    value = Variable.get("my_variable")
    
    print(f"Variable value: {value}")

my_sync_flow()

Variable.get("my_variable")

my_variable

development

production

opt/prefect/

cloudbuild.yaml

development

production

development

production

bash
     # E.g., versioning schema like 
     myimage:1.0.0-development
     myimage:1.0.0-production

/docker
      /dbt
        Dockerfile
        poetry.lock
        pyproject.toml
      /api
        Dockerfile
        poetry.lock
        pyproject.toml

yaml
    steps:
      - name: <http://gcr.io/cloud-builders/docker|gcr.io/cloud-builders/docker>
        args: ['build', '-t', '<http://gcr.io/my-project/myimage:$VERSION|gcr.io/my-project/myimage:$VERSION>', '.']
        dir: 'docker/dbt'
      - name: <http://gcr.io/cloud-builders/docker|gcr.io/cloud-builders/docker>
        args: ['tag', '<http://gcr.io/my-project/myimage:$VERSION|gcr.io/my-project/myimage:$VERSION>', '<http://gcr.io/my-project/myimage:latest|gcr.io/my-project/myimage:latest>']
    images:
      - '<http://gcr.io/my-project/myimage:$VERSION|gcr.io/my-project/myimage:$VERSION>'
      - '<http://gcr.io/my-project/myimage:latest|gcr.io/my-project/myimage:latest>'

yaml
    name: CI/CD Pipeline

    on:
      push:
        branches:
          - main
          - 'releases/**'

    jobs:
      build:
        runs-on: ubuntu-latest
        steps:
          - name: Checkout code
            uses: actions/checkout@v2

          - name: Set up Python
            uses: actions/setup-python@v2
            with:
              python-version: '3.x'

          - name: Install dependencies
            run: |
              python -m pip install --upgrade pip
              pip install -r requirements.txt

          - name: Run Tests
            run: pytest

          - name: Deploy Flow
            run: |
              python deploy_flows.py  # Customized script to deploy flows

/docker
  /base
    Dockerfile
  /dbt
    Dockerfile
    poetry.lock
    pyproject.toml
  /api
    Dockerfile
    poetry.lock
    pyproject.toml
/common-scripts
  setup_env.sh

Dockerfile
# docker/base/Dockerfile

FROM python:3.9-slim as base

# Install common dependencies
RUN apt-get update && \
    apt-get install -y --no-install-recommends \
    build-essential \
    libpq-dev \
    && rm -rf /var/lib/apt/lists/*

# Set work directory
WORKDIR /app

# Copy common scripts
COPY /common-scripts /common-scripts

# Use the common setup script
RUN chmod +x /common-scripts/setup_env.sh && /common-scripts/setup_env.sh

# Install Poetry globally
RUN pip install poetry

Dockerfile
# docker/dbt/Dockerfile

# Get common setup from the base image
FROM base as builder

# Copy project-specific files
COPY poetry.lock pyproject.toml /app/

# Install project-specific dependencies
RUN poetry config virtualenvs.create false && \
    poetry install --no-interaction --no-ansi

# Final stage with minimum runtime-only dependencies
FROM python:3.9-slim as runtime

# Copy installed dependencies from builder
COPY --from=builder /usr/local/lib/python3.9/site-packages /usr/local/lib/python3.9/site-packages
COPY --from=builder /usr/local/bin /usr/local/bin

# Set work directory
WORKDIR /app

# Copy the project files
COPY . /app

# Define the command to run your application
CMD ["dbt", "--version"]

Dockerfile
# docker/api/Dockerfile

# Get common setup from the base image
FROM base as builder

# Copy project-specific files
COPY poetry.lock pyproject.toml /app/

# Install project-specific dependencies
RUN poetry config virtualenvs.create false && \
    poetry install --no-interaction --no-ansi

# Final stage with minimum runtime-only dependencies
FROM python:3.9-slim as runtime

# Copy installed dependencies from builder
COPY --from=builder /usr/local/lib/python3.9/site-packages /usr/local/lib/python3.9/site-packages
COPY --from=builder /usr/local/bin /usr/local/bin

# Set work directory
WORKDIR /app

# Copy the project files
COPY . /app

# Define the command to run your application
CMD ["python", "api.py"]

sh
# common-scripts/setup_env.sh

#!/bin/sh

# Set environment variables
export DEBIAN_FRONTEND=noninteractive

# Common environment setup tasks
echo "Setting up common environment variables..."

sh
# Build the dbt image
docker build -f docker/dbt/Dockerfile -t myapp/dbt:latest .

# Build the api image
docker build -f docker/api/Dockerfile -t myapp/api:latest .

.deploy()

.deploy()

.deploy()

.deploy()

deployment.yaml

flow-deployment.yaml

yaml
flows:
  - name: my_flow
    schedules:
      - interval: 10 minutes
    storage:
      block: my-storage-block
    infrastructure:
      type: docker-container
      properties:
        image: my-docker-image:latest
    work_pool:
      name: my-work-pool

deploy_script.sh

sh
#!/bin/sh

# Ensure Prefect CLI is installed in your environment
pip install prefect

# Create deployment
prefect deployment create flow-deployment.yaml

# Deploy the flow
prefect deployment apply flow-deployment.yaml

docker-compose.yml

Dockerfile
# Dockerfile
FROM python:3.9-slim

# Install dependencies
COPY requirements.txt .
RUN pip install -r requirements.txt

# Copy deployment script
COPY deploy.py /deploy.py

ENTRYPOINT ["python", "/deploy.py"]

yaml
# docker-compose.yml
version: '3.8'
services:
  deploy:
    build: .
    environment:
      - PREFECT_API_KEY=${PREFECT_API_KEY}
      - PREFECT_API_URL=${PREFECT_API_URL}
    command: python deploy.py

deploy.py

python
from your_flow_module import your_flow

your_flow.deploy(
    name="your_flow_deployment_name",
    # additional params
)

sh
docker-compose up --build deploy

.deploy()

.serve()

.deploy()

.deploy()

python
  from prefect import flow
  from prefect.deployments import Deployment

  @flow
  def my_flow():
      print("Running my flow!")

  Deployment(
      flow=my_flow,
      name="my_flow_deployment",
      schedule=IntervalSchedule(interval=timedelta(minutes=10)),
      work_pool="serverless",
      tags=["production"]
  ).apply()

.serve()

.serve()

python
  import time
  from prefect import flow, serve

  @flow
  def my_flow():
      print("Running my flow!")

  if __name__ == "__main__":
      deployment = my_flow.to_deployment(name="my_flow_deployment", interval=45)
      serve(deployment)

.deploy()

.serve()

.deploy()

.serve()

.deploy()

.serve()

.deploy()

sh
    prefect block register -m prefect_aws
    prefect block register -m prefect_snowflake

python
    from prefect.blocks.system import Secret

    # Create a secret block
    Secret(value="my_secret_value").save(name="my-secret")

    # Retrieve the secret block value
    secret_block = Secret.load("my-secret")
    print(secret_block.value)  # Outputs: my_secret_value

sh
    # create secret_retrieve.py
    echo "
    from prefect.blocks.system import Secret

    secret_block = Secret.load('my-secret')
    print(secret_block.value)
    " > secret_retrieve.py

    # run the script using Prefect CLI
    prefect run -m secret_retrieve.py

python
   from prefect.blocks.system import Secret

   # Create and save a secret
   Secret(value="my_secret_value").save(name="my-secret")

python
   from prefect import flow
   from prefect.blocks.system import Secret

   @flow
   def my_flow():
       secret_block = Secret.load("my-secret")
       secret_value = secret_block.value
       print(f"The secret value is: {secret_value}")

   my_flow()

sh
   pip install --upgrade prefect

sh
   prefect cloud login

python
   from prefect.blocks.system import Secret

   try:
       secret_block = Secret.load("my-secret")
       print(secret_block.value)
   except Exception as e:
       print(f"Error retrieving secret: {e}")

sh
   prefect block list
   # Look for your secret block in the output

python
   from prefect import flow, get_run_logger
   from prefect.blocks.system import Secret

   @flow
   def my_flow():
       logger = get_run_logger()
       try:
           secret_block = Secret.load("my-secret")
           secret_value = secret_block.value
           <http://logger.info|logger.info>(f"The secret value is: {secret_value}")
       except Exception as e:
           logger.error(f"Error retrieving secret: {e}")

   my_flow()

coroutine

async

await

python
from prefect import flow, get_run_logger
from prefect.blocks.system import Secret

@flow
async def my_flow():
    logger = get_run_logger()
    try:
        secret_block = await Secret.load("my-secret")
        secret_value = secret_block.value
        <http://logger.info|logger.info>(f"The secret value is: {secret_value}")
    except Exception as e:
        logger.error(f"Error retrieving secret: {e}")

my_flow()

async def

await

Secret.load

python
    from prefect.blocks.system import Secret

    # Create and save a secret
    secret = Secret(value="my_secret_value")
    secret.save(name="my-secret")

python
    from prefect import flow, get_run_logger
    from prefect.blocks.system import Secret

    @flow
    async def my_flow():
        logger = get_run_logger()
        try:
            # Await the loading of the secret block
            secret_block = await Secret.load("my-secret")
            secret_value = secret_block.value
            <http://logger.info|logger.info>(f"The secret value is: {secret_value}")
        except Exception as e:
            logger.error(f"Error retrieving secret: {e}")

    if __name__ == "__main__":
        import anyio
        anyio.run(my_flow)

anyio

sh
    pip install anyio

async

await

coroutine

Secret.load

python
# flows.py
from prefect import flow, get_run_logger

@flow
def my_flow(param1, param2):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Running with param1={param1} and param2={param2}")
    return param1 + param2

python
# deploy.py
from prefect.deployments import Deployment
from prefect.server.schemas.schedules import IntervalSchedule
from datetime import timedelta
from flows import my_flow

def create_deployment():
    schedule = IntervalSchedule(interval=timedelta(minutes=10))
    deployment = Deployment(
        flow=my_flow,
        name="my_flow_deployment",
        schedule=schedule,
        parameters={"param1": 1, "param2": 2},
        tags=["development"],
    )
    deployment.apply()

if __name__ == "__main__":
    create_deployment()

python
# test_flow.py
from flows import my_flow

def test_my_flow():
    result = my_flow(param1=1, param2=2)
    assert result == 3

if __name__ == "__main__":
    test_my_flow()
    print("All tests passed!")

sh
    python test_flow.py

sh
    # Register and run
    prefect deployment build -n my_flow_deployment deploy.py:my_flow -p "param1=1 param2=2"
    prefect deployment apply my_flow-deployment.yaml
    prefect deployment run 'my_flow_deployment/my_flow'

unittest.mock

python
    # example_mock.py
    from unittest.mock import patch
    from flows import my_flow

    @patch("flows.some_external_service")
    def test_my_flow(mock_service):
        mock_service.return_value = "mocked_value"
        result = my_flow(param1=1, param2=2)
        assert result == 3

    if __name__ == "__main__":
        test_my_flow()
        print("All mocked tests passed!")

sh
    # Start a local agent
    prefect agent start -q 'default'
    # Apply and run the deployment locally
    prefect deployment run -n 'my_flow_deployment'

main