LinkedIn
Twitter
I'm not that proficient with Python coding, so I've asked AI for little help. However
main.py
from googleapiclient.discovery import build
from google.oauth2 import service_account
from google.cloud import storage
from google.cloud import secretmanager
from googleapiclient.http import MediaIoBaseDownload
import io
import os
import json
FOLDER_ID = os.environ.get('FOLDER_ID')
BUCKET_NAME = os.environ.get('BUCKET_NAME')
SECRET_ID = os.environ.get('SECRET_ID')
PROJECT_ID = os.environ.get('PROJECT_ID')
def access_secret_version(project_id, secret_id, version_id='latest'):
client = secretmanager.SecretManagerServiceClient()
name = f"projects/{project_id}/secrets/{secret_id}/versions/{version_id}"
try:
response = client.access_secret_version(request={"name": name})
payload = response.payload.data.decode("UTF-8")
print(f"Successfully accessed secret: {secret_id}")
return payload
except Exception as e:
print(f"Error accessing secret version: {e}")
raise e
def get_service_account_credentials(secret_payload):
try:
info = json.loads(secret_payload)
credentials = service_account.Credentials.from_service_account_info(info)
print("Service account credentials successfully created.")
return credentials
except Exception as e:
print(f"Error creating service account credentials: {e}")
raise e
def list_files(service, folder_id):
try:
results = service.files().list(
q=f"'{folder_id}' in parents and mimeType != 'application/vnd.google-apps.folder'",
pageSize=1000, fields="nextPageToken, files(id, name)"
).execute()
items = results.get('files', [])
print(f"Successfully listed files in folder: {folder_id}. Found {len(items)} files.")
return items
except Exception as e:
print(f"Error listing files: {e}")
raise e
def download_file(service, file_id, file_name):
request = service.files().get_media(fileId=file_id)
fh = io.BytesIO()
downloader = MediaIoBaseDownload(fh, request)
done = False
while not done:
try:
status, done = downloader.next_chunk()
print(f"Download {int(status.progress() * 100)}% complete for file {file_name}.")
except Exception as e:
print(f"Error downloading file {file_name}: {e}")
raise e
fh.seek(0)
print(f"File {file_name} downloaded successfully.")
return fh
def upload_to_gcs(file_stream, bucket_name, object_name):
try:
storage_client = storage.Client()
bucket = storage_client.bucket(bucket_name)
blob = bucket.blob(object_name)
blob.upload_from_file(file_stream, rewind=True)
print(f"{object_name} uploaded to {bucket_name}.")
except Exception as e:
print(f"Error uploading to GCS: {e}")
raise e
def sync_drive_to_gcs(request):
try:
secret_payload = access_secret_version(PROJECT_ID, SECRET_ID)
credentials = get_service_account_credentials(secret_payload)
service = build('drive', 'v3', credentials=credentials)
items = list_files(service, FOLDER_ID)
for item in items:
file_id = item['id']
file_name = item['name']
print(f"Downloading {file_name} from Google Drive")
file_stream = download_file(service, file_id, file_name)
print(f"Uploading {file_name} to GCS")
upload_to_gcs(file_stream, BUCKET_NAME, file_name)
return 'Sync complete', 200
except Exception as e:
print(f"Error in sync_drive_to_gcs: {e}")
return f"Internal Server Error: {e}", 500
requirements.txt
google-cloud-secret-manager
google-api-python-client
google-auth-httplib2
google-auth-oauthlib
google-cloud-storage
Function variables
1. Create Service Account and create service account key.
2. Create entry at Secret Manager to store key content
3. Grant IAM role Secret Manager Secret Accessor to your newly created Service Account
4. Enable APIs: Google Drive API
5. Create bucket
6. Obtain Service Account email address and Share Google Drive folder with that Service Account ( Go to Google Drive - Folder -> Share -> Copy/paste email address -> Set to Editor -> Share)
7. Create Cloud Function. If you get question about API , click Enable APIs.
8. Name your function -> Allow unauthenticated invocations ( only for testing !)
9. Choose your service account created at the beginning as "Runtime service account"
10. Add Runtime Variables with values
FOLDER_ID = FOlder ID taken from Google Drive
BUCKET_NAME = name of the bucket where files will be copy/pasted
SECRET_ID = name of secret, under where you keeping ServiceAccount Key content
PROJECT_ID = ID of the project where cloud function will be created
EXAMPLE:
11. Choose Python 3.10 as runtime, sync_drive_to_gcs as Entry POint.
Once Function deployed, you will see Function URL, once clicked Function will be triggered.
Output from logs:
IMPORTANT: If you have big files, like movies or something, you must create cloud function with big amount of Memory.
Important2: If you have any spreadsheet, docs, PPTs or something, you will get following error:
Error downloading file Untitled spreadsheet: <HttpError 403 when requesting https://www.googleapis.com/drive/v3/files/1TiaGARyv4IdFuh2jcDgQBJERBVNdcJDWBVfFfEGziVQ?alt=media returned "Only files with binary content can be downloaded. Use Export with Docs Editors files.". Details: "[{'message': 'Only files with binary content can be downloaded. Use Export with Docs Editors files.', 'domain': 'global', 'reason': 'fileNotDownloadable', 'location': 'alt', 'locationType': 'parameter'}]">
