gsutil

import subprocess import os # Ensure gcloud is authenticated (e.g., via `gcloud auth login` in terminal) # For programmatic use, consider setting GOOGLE_APPLICATION_CREDENTIALS # os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = '/path/to/keyfile.json' bucket_name = os.environ.get('GSUTIL_TEST_BUCKET', 'my-test-bucket-12345') file_name = 'test_file.txt' file_content = 'Hello, Cloud Storage!' # Create a local file to upload with open(file_name, 'w') as f: f.write(file_content) try: # Create a bucket print(f"Creating bucket gs://{bucket_name}...") subprocess.run(['gsutil', 'mb', f'gs://{bucket_name}'], check=True, capture_output=True) print(f"Bucket gs://{bucket_name} created.") # Upload a file print(f"Uploading {file_name} to gs://{bucket_name}/{file_name}...") subprocess.run(['gsutil', 'cp', file_name, f'gs://{bucket_name}/{file_name}'], check=True, capture_output=True) print("File uploaded.") # List bucket contents print(f"Listing contents of gs://{bucket_name}:") result = subprocess.run(['gsutil', 'ls', f'gs://{bucket_name}'], check=True, capture_output=True, text=True) print(result.stdout) # Download the file downloaded_file_name = 'downloaded_test_file.txt' print(f"Downloading gs://{bucket_name}/{file_name} to {downloaded_file_name}...") subprocess.run(['gsutil', 'cp', f'gs://{bucket_name}/{file_name}', downloaded_file_name], check=True, capture_output=True) print("File downloaded.") with open(downloaded_file_name, 'r') as f: content = f.read() print(f"Downloaded content: {content}") except subprocess.CalledProcessError as e: print(f"gsutil command failed: {e}") print(f"STDOUT: {e.stdout.decode()}") print(f"STDERR: {e.stderr.decode()}") finally: # Clean up (delete file and bucket) print(f"Cleaning up: deleting objects from gs://{bucket_name}...") subprocess.run(['gsutil', '-m', 'rm', '-r', f'gs://{bucket_name}/**'], check=False, capture_output=True) print(f"Deleting bucket gs://{bucket_name}...") subprocess.run(['gsutil', 'rb', f'gs://{bucket_name}'], check=False, capture_output=True) print("Cleanup complete.") if os.path.exists(file_name): os.remove(file_name) if os.path.exists(downloaded_file_name): os.remove(downloaded_file_name)