|
from google.oauth2 import service_account |
|
from googleapiclient.discovery import build |
|
from googleapiclient.http import MediaIoBaseDownload |
|
import io |
|
import os |
|
|
|
|
|
def download_file(file_id, file_name): |
|
|
|
creds_info = { |
|
"type": "service_account", |
|
"project_id": os.environ.get('GCP_PROJECT_ID'), |
|
"private_key_id": os.environ.get('GCP_PRIVATE_KEY_ID'), |
|
"private_key": os.environ.get('GCP_PRIVATE_KEY').replace('\\n', '\n'), |
|
"client_email": os.environ.get('GCP_CLIENT_EMAIL'), |
|
"client_id": os.environ.get('GCP_CLIENT_ID'), |
|
"auth_uri": "https://accounts.google.com/o/oauth2/auth", |
|
"token_uri": "https://oauth2.googleapis.com/token", |
|
"auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs", |
|
"client_x509_cert_url": os.environ.get('GCP_CLIENT_X509_CERT_URL'), |
|
"universe_domain": "googleapis.com" |
|
} |
|
|
|
creds = service_account.Credentials.from_service_account_info(creds_info) |
|
|
|
|
|
drive_service = build('drive', 'v3', credentials=creds) |
|
|
|
request = drive_service.files().get_media(fileId=file_id) |
|
fh = io.FileIO(file_name, 'wb') |
|
downloader = MediaIoBaseDownload(fh, request) |
|
done = False |
|
while done is False: |
|
status, done = downloader.next_chunk() |
|
print(f'Download {int(status.progress() * 100)}%.') |
|
|
|
|
|
if __name__ == '__main__': |
|
|
|
download_file(os.environ.get("CODE_URL"), 'code.zip') |
|
download_file(os.environ.get("CODE_URL2"), 'code2.zip') |
|
|
|
|