Simple repository with GitHub Actions workflow - manually specified flows in the GHA matrix + explicitly defined S3 FileStorageBlock

DataOps Tutorials for Prefect 2.0
, , ,
#1

GitHub Actions workflow

name: CI
on:
  push:
    branches: [ main ]

jobs:
  build:
    runs-on: ubuntu-latest
    env:
      API_KEY: ${{ secrets.PREFECT_API_KEY}}
    strategy:
      matrix:
        flows: [ hello_flow.py, hello_flow_2.py ]
    steps:
      - uses: actions/checkout@v3
      - name: Set up Python 3.9
        uses: actions/setup-python@v3
        with:
          python-version: 3.9
      - name: Dev dependencies
        run: |
          pip install s3fs
          pip install -U "prefect>=2.0b"
      - name: Login to Prefect Cloud 2.0
        run: prefect cloud login --key $API_KEY --workspace 'annaprefect/prod'
      - name: Validate Prefect version
        run: prefect version
      - name: Configure AWS Credentials
        uses: aws-actions/configure-aws-credentials@v1
        with:
          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          aws-region: us-east-1
      - name: Create deployment
        run: cd flows && prefect deployment create ${{ matrix.flows }}

Prefect flow

Example flows/hello_flow.py:

import platform
from prefect import task, flow
from prefect import get_run_logger
from prefect.deployments import DeploymentSpec
from prefect.blocks.storage import FileStorageBlock


@task
def say_hi():
    logger = get_run_logger()
    logger.info("Hello world!")


@task
def print_platform_info():
    logger = get_run_logger()
    logger.info(
        "Platform information: IP = %s, Python = %s, Platform type = %s, System Version = %s",
        platform.node(),
        platform.python_version(),
        platform.platform(),
        platform.version(),
    )


@flow
def hello_flow():
    hi = say_hi()
    print_platform_info(wait_for=[hi])


DeploymentSpec(
    name="dev",
    flow_location="hello_flow.py",
    flow_storage=FileStorageBlock(base_path="s3://prefect-orion/flows"),
)


if __name__ == "__main__":
    hello_flow()

And another one to test multiple flows in the same repo - flows/hello_flow_2.py:

import platform
from prefect import task, flow
from prefect import get_run_logger
from prefect.deployments import DeploymentSpec
from prefect.blocks.storage import FileStorageBlock


@task
def say_hi():
    logger = get_run_logger()
    logger.info("Hello world!")


@flow
def hello_flow_2():
    hi = say_hi()


DeploymentSpec(
    name="dev",
    flow_location="hello_flow_2.py",
    flow_storage=FileStorageBlock(base_path="s3://prefect-orion/flows"),
)

if __name__ == "__main__":
    hello_flow_2()
CI/CD for Prefect 2.0