73 lines
2.6 KiB
Python
73 lines
2.6 KiB
Python
from google.oauth2.credentials import Credentials
|
|
from google_auth_oauthlib.flow import Flow
|
|
from googleapiclient.discovery import build
|
|
from googleapiclient.http import MediaFileUpload, MediaIoBaseDownload
|
|
import os
|
|
import json
|
|
|
|
# This file should be downloaded from Google Cloud Console
|
|
CLIENT_SECRETS_FILE = "credentials.json"
|
|
SCOPES = ["https://www.googleapis.com/auth/drive.file"]
|
|
|
|
def get_drive_service(credentials_info=None):
|
|
if isinstance(credentials_info, str):
|
|
credentials_info = json.loads(credentials_info)
|
|
|
|
creds = None
|
|
if credentials_info:
|
|
creds = Credentials.from_authorized_user_info(credentials_info, SCOPES)
|
|
|
|
if not creds or not creds.valid:
|
|
# This should not happen in a web flow if the user is authenticated.
|
|
# If it does, it means the credentials have expired or are invalid,
|
|
# and the user needs to re-authenticate.
|
|
# The API should handle this by returning a 401 Unauthorized error.
|
|
return None
|
|
|
|
return build('drive', 'v3', credentials=creds)
|
|
|
|
def upload_file(drive_service, file_path, folder_id=None):
|
|
file_metadata = {
|
|
'name': os.path.basename(file_path)
|
|
}
|
|
if folder_id:
|
|
file_metadata['parents'] = [folder_id]
|
|
|
|
media = MediaFileUpload(file_path, resumable=True)
|
|
file = drive_service.files().create(
|
|
body=file_metadata,
|
|
media_body=media,
|
|
fields='id'
|
|
).execute()
|
|
|
|
return file.get('id')
|
|
|
|
def download_file(drive_service, file_id, destination):
|
|
request = drive_service.files().get_media(fileId=file_id)
|
|
with open(destination, "wb") as fh:
|
|
downloader = MediaIoBaseDownload(fh, request)
|
|
done = False
|
|
while done is False:
|
|
status, done = downloader.next_chunk()
|
|
print(f"Download {int(status.progress() * 100)}%.")
|
|
|
|
def authenticate(redirect_uri):
|
|
flow = Flow.from_client_secrets_file(
|
|
CLIENT_SECRETS_FILE, SCOPES, redirect_uri=redirect_uri
|
|
)
|
|
authorization_url, state = flow.authorization_url(
|
|
access_type='offline', include_granted_scopes='true'
|
|
)
|
|
# Store state in session or a temporary store to verify in the callback
|
|
# For now, we'll just return it. A real app should handle this securely.
|
|
return authorization_url, state
|
|
|
|
def exchange_code_for_credentials(code, redirect_uri, state):
|
|
flow = Flow.from_client_secrets_file(
|
|
CLIENT_SECRETS_FILE, SCOPES, redirect_uri=redirect_uri
|
|
)
|
|
# The state parameter should be verified against the value from the auth request
|
|
flow.fetch_token(code=code)
|
|
credentials = flow.credentials
|
|
return credentials
|