Simple repository with GitHub Actions workflow for multiple flows/deployments (automatically inferred flows in the GHA matrix) and S3 storage

DataOps Tutorials for Prefect 2.0
#1

GitHub Actions workflow

name: CI
on:
  push:
    branches: [ 04-s3-fsspec-blocks-dynamic-matrix ]

  # Allows you to run this workflow manually from the Actions tab
  workflow_dispatch:

jobs:
  list-flows:
    runs-on: ubuntu-latest
    outputs:
      matrix: ${{ steps.set-matrix.outputs.matrix }}
    steps:
      - uses: actions/checkout@v3
      - id: set-matrix
        run: echo "::set-output name=matrix::$(ls flows/*.py | jq -R -s -c 'split("\n")[:-1]')"
  build:
    needs: list-flows
    runs-on: ubuntu-latest
    env:
      API_KEY: ${{ secrets.PREFECT_API_KEY}}
    strategy:
      matrix:
        flows: ${{ fromJson(needs.list-flows.outputs.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: prefect deployment create ${{ matrix.flows }}

Prefect flow

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=hello_flow,
    tags=["local"],
    flow_storage=FileStorageBlock(base_path="s3://prefect-orion/flows"),
)


if __name__ == "__main__":
    hello_flow()