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