prefect-gcp

prefect block register -m prefect_gcp

from prefect_gcp import GcpCredentials

# replace this PLACEHOLDER dict with your own service account info
service_account_info = {
  "type": "service_account",
  "project_id": "PROJECT_ID",
  "private_key_id": "KEY_ID",
  "private_key": "-----BEGIN PRIVATE KEY-----\nPRIVATE_KEY\n-----END PRIVATE KEY-----\n",
  "client_email": "SERVICE_ACCOUNT_EMAIL",
  "client_id": "CLIENT_ID",
  "auth_uri": "https://accounts.google.com/o/oauth2/auth",
  "token_uri": "https://accounts.google.com/o/oauth2/token",
  "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
  "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/SERVICE_ACCOUNT_EMAIL"
}

GcpCredentials(
    service_account_info=service_account_info
).save("CREDENTIALS-BLOCK-NAME")

from prefect_gcp.bigquery import GcpCredentials, BigQueryWarehouse

gcp_credentials = GcpCredentials.load("CREDENTIALS-BLOCK-NAME")

bigquery_block = BigQueryWarehouse(
    gcp_credentials = gcp_credentials,
    fetch_size = 1  # Optional: specify a default number of rows to fetch when calling fetch_many
)
bigquery_block.save("BIGQUERY-BLOCK-NAME")

from prefect_gcp import GcpCredentials, GcpSecret
gcp_credentials = GcpCredentials.load("CREDENTIALS-BLOCK-NAME")

gcp_secret = GcpSecret(
    secret_name = "your-secret-name",
    secret_version = "latest",
    gcp_credentials = gcp_credentials
)

gcp_secret.save("SECRET-BLOCK-NAME")

from prefect_gcp import GcpCredentials, GcsBucket

gcs_bucket = GcsBucket(
    bucket="BUCKET-NAME",
    gcp_credentials=GcpCredentials.load("BIGQUERY-BLOCK-NAME")
)
gcs_bucket.save("GCS-BLOCK-NAME")

from prefect import flow
from prefect_gcp.bigquery import GcpCredentials, BigQueryWarehouse

@flow
def bigquery_flow():
    all_rows = []
    gcp_credentials = GcpCredentials.load("CREDENTIALS-BLOCK-NAME")

    client = gcp_credentials.get_bigquery_client()
    client.create_dataset("test_example", exists_ok=True)

    with BigQueryWarehouse(gcp_credentials=gcp_credentials) as warehouse:
        warehouse.execute(
            "CREATE TABLE IF NOT EXISTS test_example.customers (name STRING, address STRING);"
        )
        warehouse.execute_many(
            "INSERT INTO test_example.customers (name, address) VALUES (%(name)s, %(address)s);",
            seq_of_parameters=[
                {"name": "Marvin", "address": "Highway 42"},
                {"name": "Ford", "address": "Highway 42"},
                {"name": "Unknown", "address": "Highway 42"},
            ],
        )
        while True:
            # Repeated fetch* calls using the same operation will
            # skip re-executing and instead return the next set of results
            new_rows = warehouse.fetch_many("SELECT * FROM test_example.customers", size=2)
            if len(new_rows) == 0:
                break
            all_rows.extend(new_rows)
    return all_rows


if __name__ == "__main__":
    bigquery_flow()

from pathlib import Path
from prefect import flow
from prefect_gcp import GcpCredentials, GcsBucket


@flow
def cloud_storage_flow():
    # create a dummy file to upload
    file_path = Path("test-example.txt")
    file_path.write_text("Hello, Prefect!")

    gcp_credentials = GcpCredentials.load("CREDENTIALS-BLOCK-NAME")
    gcs_bucket = GcsBucket(
        bucket="BUCKET-NAME",
        gcp_credentials=gcp_credentials
    )

    gcs_bucket_path = gcs_bucket.upload_from_path(file_path)
    downloaded_file_path = gcs_bucket.download_object_to_path(
        gcs_bucket_path, "downloaded-test-example.txt"
    )
    return downloaded_file_path.read_text()


if __name__ == "__main__":
    cloud_storage_flow()

from prefect import flow
from prefect_gcp import GcpCredentials, GcpSecret


@flow
def secret_manager_flow():
    gcp_credentials = GcpCredentials.load("CREDENTIALS-BLOCK-NAME")
    gcp_secret = GcpSecret(secret_name="test-example", gcp_credentials=gcp_credentials)
    gcp_secret.write_secret(secret_data=b"Hello, Prefect!")
    secret_data = gcp_secret.read_secret()
    gcp_secret.delete_secret()
    return secret_data


if __name__ == "__main__":
    secret_manager_flow()