from google.oauth2 import service_account as sa
from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload, MediaIoBaseUpload
from .abstract_chart_deployer import AbstractChartDeployer
from io import BytesIO
import os
[docs]
class GdriveChartDeployer(AbstractChartDeployer):
DEFAULT_SA_PATH = "./service_account.json"
DEFAULT_SCOPES = ['https://www.googleapis.com/auth/drive']
def __init__(self, folder_id, mime_type=None, sa_path=None, params=None):
self.sa_path = sa_path or self.DEFAULT_SA_PATH
self.auth = sa.Credentials.from_service_account_file(
self.sa_path,
scopes=self.DEFAULT_SCOPES
)
self.drive_service = build('drive', 'v3', credentials=self.auth)
self.folder_id = folder_id
self.mime_type = mime_type
[docs]
def deploy(self, fp, filename):
# # THE FOLLOWING CODE WOULD CHECK FOR EXISTING FILES WITH THE SAME FILENAME
# # COMMENTED OUT WHILE WE HAVE NO SYSTEM TO NAME FILES
# query = f"name = '{self.file_name}' and '{self.folder_id}' in parents and trashed = false"
# results = self.drive_service.files().list(
# q=query,
# fields="files(id, name)"
# ).execute()
# files = results.get('files', [])
files = []
file_metadata = {
'name': filename,
'parents': [self.folder_id],
}
if hasattr(fp, 'getvalue'):
content = BytesIO(fp.getvalue().encode("utf-8"))
elif isinstance(fp, (str, bytes, os.PathLike)):
with open(fp, 'rb') as file:
content = file.read()
else:
raise TypeError("fp must be a file-like object or a file path")
if isinstance(content, BytesIO):
media = MediaIoBaseUpload(content, mimetype=self.mime_type, resumable=True)
else:
media = MediaFileUpload(fp, mimetype=self.mime_type)
if files:
request = self.drive_service.files().update(
fileId=files[0].get('id'),
body={'name': self.file_name},
media_body=media,
supportsAllDrives=True
)
else:
request = self.drive_service.files().create(
body=file_metadata,
media_body=media,
fields='id'
)
response = request.execute()
return response.get('id')