CI/CD with Cloud Build for Prefect

README.md

project
└───custom-builder-prefect
│   │   Dockerfile
|
└───prefect-code
    └───flows
    |   │   first_flow.py
    |   │   second_flow.py
    |
    └───ci-cd
        │   cloudbuild.yaml

0. Set up GCP

• Create a new project
• Install gcloud CLI
• Authenticate with gcloud
• Enable cloud build API

1. Submit Build of GCR Image

cd ./customer-builder-prefect gcloud builds submit --tag gcr.io/{project_name}/{image_name} cd ..

2. Install Cloud Build Local

sudo apt-get install google-cloud-sdk-cloud-build-local cloud-build-local --version

3. Run Local Build

cd ./prefect-code cloud-build-local --config=ci-cd/cloudbuild.yaml --dryrun=false .

cloudbuild.yaml

steps:
- id: 'Generate full list of flows/DeploymentSpecs'
  name: 'ubuntu'
  entrypoint: 'printf'
  args:
    - "'%s\n' workspace/flows/* > flow_list.txt"

- id: 'Login to Prefect Cloud 2.0'
  name: 'gcr.io/{project_name}/{image_name}'
  args: ['prefect', 'cloud', 'login', '--key', '{insert_key}', '--workspace', '{insert_workspace}']
  
  # The next step will be to run the deployments once I get the above working

Dockerfile

FROM prefecthq/prefect:2.0b6-python3.9

first_flow.py

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


@task
def set_parameter():
    print("This doesn't do much")
    return 42


@task
def print_parameter(parameter: str):
    print(parameter)


@flow
def main_flow():
    param = set_parameter()
    print_parameter(param)
    return

DeploymentSpec(
    name="first-deployment",
    flow=main_flow,
    tags=["dev"],
    flow_storage=FileStorageBlock(base_path="gcs://{bucket_name}/flows"),
)

if __name__ == "__main__":
    main_flow()

second_flow.py

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


@task
def set_parameter():
    print("This doesn't do much either")
    return 42


@task
def print_parameter(parameter: str):
    print(parameter)


@flow
def main_flow():
    param = set_parameter()
    print_parameter(param)
    return

DeploymentSpec(
    name="second-deployment",
    flow=main_flow,
    tags=["dev"],
    flow_storage=FileStorageBlock(base_path="gcs://{bucket_name}/flows"),
)

if __name__ == "__main__":
    main_flow()