gcs-oauth2-boto-plugin

import os import gcs_oauth2_boto_plugin from boto.gs.connection import GSConnection from boto.exception import GSResponseError # NOTE: This library is DEPRECATED. Use `google-cloud-storage` instead. # This example assumes you have authenticated via: # `gcloud auth application-default login` # or have GOOGLE_APPLICATION_CREDENTIALS pointing to a service account key. bucket_name = os.environ.get('GCS_BOTO_EXAMPLE_BUCKET', 'your-gcs-boto-example-bucket') print(f"Attempting to connect to GCS using boto with gcs-oauth2-boto-plugin...") try: # After importing gcs_oauth2_boto_plugin, boto will automatically use it. conn = GSConnection() # Check if the bucket exists or create it try: bucket = conn.get_bucket(bucket_name) print(f"Successfully connected to bucket '{bucket_name}'.") except GSResponseError as e: if e.status == 404: print(f"Bucket '{bucket_name}' not found, attempting to create...") bucket = conn.create_bucket(bucket_name) print(f"Bucket '{bucket_name}' created successfully.") else: raise # Upload a simple string to the bucket key_name = 'hello_boto_gcs.txt' key = bucket.new_key(key_name) key.set_contents_from_string('Hello from gcs-oauth2-boto-plugin!') print(f"Uploaded '{key_name}' to '{bucket_name}'.") # Download and print the content key_retrieved = bucket.get_key(key_name) if key_retrieved: content = key_retrieved.get_contents_as_string().decode('utf-8') print(f"Content of '{key_name}': {content}") else: print(f"Failed to retrieve '{key_name}'.") # Clean up (optional) # key_retrieved.delete() # print(f"Deleted '{key_name}'.") except Exception as e: print(f"An error occurred: {e}") print("Please ensure you have authenticated to Google Cloud (e.g., `gcloud auth application-default login`) and have correct permissions.")