google_resumable_media_gcs_upload_with_authorized_transport.py

1import io
2import google.auth
3import google.auth.transport.requests
4from google.resumable_media import requests
5
6# 1. Get credentials and create an authorized transport (Requests)
7credentials, project_id = google.auth.default()
8transport = google.auth.transport.requests.AuthorizedSession(credentials)
9
10# 2. Define the upload details
11bucket = "my-bucket"
12object_name = "my-file.txt"
13upload_url = f"https://www.googleapis.com/upload/storage/v1/b/{bucket}/o?uploadType=resumable"
14
15# 3. Create the data to be uploaded
16stream = io.BytesIO(b"This is the content of the file.")
17content_type = "text/plain"
18
19# 4. Initialize the upload (this returns a resumable session URL)
20upload = requests.ResumableUpload(upload_url, chunk_size=1024 * 1024)
21upload.initiate(
22    transport,
23    stream,
24    {"name": object_name},
25    content_type,
26)
27
28# 5. Perform the upload
29response = upload.transmit_next_chunk(transport)
30
31print(f"Upload Status Code: {response.status_code}")