Simple repository with GitHub Actions workflow for multiple flows/deployments (automatically inferred flows in the GHA matrix) and GCS storage (with both fsspec gcsfs + GoogleCloudStorageBlock examples)

GitHub Actions workflow

name: CI
on:
  push:
    branches: [ 05-fsspec-gcs-dynamic-matrix ]
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]')"
  deploy:
    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 gcsfs
          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
      # Copy and paste the contents of the Service Account JSON file into the secret https://github.com/marketplace/actions/authenticate-to-google-cloud#authenticating-via-service-account-key-json-1
      - name: Authenticate to GCP
        uses: 'google-github-actions/auth@v0.7.1'
        with:
          credentials_json: '${{ secrets.GOOGLE_APPLICATION_CREDENTIALS }}'
          create_credentials_file: true
      - name: Create deployment
        run: prefect deployment create ${{ matrix.flows }}

Or using keyless authentication:

name: CI
on:
  push:
    branches: [ main ]
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]')"
  deploy:
    needs: list-flows
    runs-on: ubuntu-latest
    permissions:
      contents: read
      id-token: write
    strategy:
      matrix:
        flows: ${{ fromJson(needs.list-flows.outputs.matrix) }}
    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 gcsfs
          pip install -U "prefect>=2.0b"
      - name: Login to Prefect Cloud 2.0
        run: prefect cloud login --key ${{ secrets.PREFECT_API_KEY }} --workspace 'danilodrobacnzytecouk/nzyte-prefect'
      - name: Validate Prefect version
        run: prefect version
      - name: Authenticate to GCP
        uses: 'google-github-actions/auth@v0.7.1'
        with:
          workload_identity_provider: 'projects/170961541935/locations/global/workloadIdentityPools/github-actions-pool/providers/github-actions-provider'
          service_account: 'prefect-service-account@nzyte-modern-data-stack.iam.gserviceaccount.com'
      - 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_2():
    hi = say_hi()
    print_platform_info(wait_for=[hi])


DeploymentSpec(
    name="gcs",
    flow=hello_flow_2,
    tags=["local"],
    flow_storage=FileStorageBlock(base_path="gcs://prefect-orion/flows"),
)

if __name__ == "__main__":
    hello_flow_2()

And using GoogleCloudStorageBlock:

import os
import json
import io
import platform
from prefect import task, flow, __version__
from prefect import get_run_logger
from prefect.deployments import DeploymentSpec
from prefect.blocks.storage import FileStorageBlock, GoogleCloudStorageBlock


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


@task
def print_platform_info():
    logger = get_run_logger()
    logger.info(__version__)
    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])

sa_file = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS")

with io.open(sa_file, "r", encoding="utf-8") as json_file:
    service_account_info = json.load(json_file)


DeploymentSpec(
    name="gcs",
    flow=hello_flow,
    tags=["local"],
    flow_storage=GoogleCloudStorageBlock(
        bucket="prefect-orion",
        service_account_info=service_account_info
    ),
)


if __name__ == "__main__":
    hello_flow()