[Python] google.auth.exceptions.RefreshError: ('No access token in response.', {'id_token':

I am trying to connect to google drive to perform some operations in drive folder. I built a function and it worked. but when I tried implementing OOP and defined classes, I am getting this error. I have checked all the env and config variables and they are correct. Find below my codes in FP and OOP

function: Working


from google.oauth2 import service_account
from googleapiclient.discovery import build

SCOPES = ['https://www.googleapis.com/auth/drive']
SERVICE_ACCOUNT_FILE = 'c4s-drive-secretkey.json'

credentials = service_account.Credentials.from_service_account_file(
SERVICE_ACCOUNT_FILE,
scopes=SCOPES
)


service = build('drive', 'v3', credentials=credentials)

query = f"'1Qe***************MzGVwQ-' in parents and trashed = false"
results = service.files().list(q=query,

fields="files(id, name)",
supportsAllDrives=True,
includeItemsFromAllDrives=True).execute()
files = results.get('files', [])
if files:
for file in files:
print(f"Deleting {file['name']} ({file['id']})")
service.files().delete(fileId=file['id'],supportsAllDrives=True).execute()


OOP: Not working

Load.py

class DriveLoader:
def __init__(self, PARENT_FOLDER_ID,download_path,SCOPES,SERVICE_ACC_FILE):
self.SCOPES = SCOPES
self.SERVICE_ACC_FILE = SERVICE_ACC_FILE
self.PARENT_FOLDER_ID = PARENT_FOLDER_ID
self.download_path = download_path
creds = service_account.Credentials.from_service_account_file(self.SERVICE_ACC_FILE, scopes = self.SCOPES)
self.service = build('drive', 'v3', credentials=creds)

def clear_folder(self):
query = f"'{self.PARENT_FOLDER_ID}' in parents and trashed = false"
results = self.service.files().list(q=query,
fields="files(id, name)",
supportsAllDrives=True,
includeItemsFromAllDrives=True).execute()

files = results.get('files', [])

if files:
for file in files:
print(f"Deleting {file['name']} ({file['id']})")
self.service.files().delete(fileId=file['id'],supportsAllDrives=True).execute()



Script.py:

class ExtractLoad:

def __init__(self, url, web_username, web_pass, form_lst, download_path, suffix,PARENT_FOLDER_ID,SCOPE,SERVICE_ACC_FILE):
self.web_username = web_username
self.web_pass = web_pass
self.download_path = download_path
self.form_lst = form_lst
self.suffix = suffix
self.url = url
self.PARENT_FOLDER_ID = PARENT_FOLDER_ID
self.SCOPE = SCOPE
self.SERVICE_ACC_FILE = SERVICE_ACC_FILE
def drive_loader(self):

loader = DriveLoader(self.PARENT_FOLDER_ID,self.download_path,self.SCOPE,self.SERVICE_ACC_FILE)
loader.clear_folder()



Script.py:

class C4SExecutor:
def __init__(self):
load_dotenv()
self.ENV = dotenv_values(".env")
with open("config.toml", "r") as file:
self.config = toml.load(file)

def run_pipeline(self):
for form in self.config["FORMS"]:
pipeline = ExtractLoad(form["url"], self.ENV["web_username"], self.ENV["web_pass"], form["forms"], form["download_path"], form["suffix"],form["PARENT_FOLDER_ID"], self.ENV["SCOPE"], self.ENV["SERVICE_ACC_FILE"])

pipeline.drive_loader()
logger.info("screening files uploaded to drive")


main.py:

from script import C4SExecutor

executor = C4SExecutor()
executor.run_pipeline()

Continue reading...