prefect_gcp_gcs_bucket_file_upload_quickstart.py

1from prefect import flow, task
2from prefect_gcp.cloud_storage import GcsBucket
3
4@task
5def create_content():
6    return "Hello from Prefect and GCP!"
7
8@flow
9def gcs_upload_flow(bucket_block_name: str):
10    # Load the pre-configured GCS Bucket block
11    gcs_bucket = GcsBucket.load(bucket_block_name)
12    
13    # Create some content to upload
14    content = create_content()
15    
16    # Upload the content to the bucket
17    gcs_bucket.upload_from_path(
18        path="example_file.txt",
19        from_path=None  # This can be used if uploading a local file
20    )
21    
22    # Alternatively, writing a string directly
23    gcs_bucket.write_path("hello_gcp.txt", content.encode())
24
25if __name__ == "__main__":
26    # Ensure you have created a GCS Bucket block named 'my-gcs-bucket' 
27    # in the Prefect UI or via CLI before running
28    gcs_upload_flow(bucket_block_name="my-gcs-bucket")