Change google adapter to use service account.

This commit is contained in:
redxef 2022-10-13 00:47:08 +02:00
parent e683a90b34
commit 21be48f8e3
Signed by: redxef
GPG key ID: 7DAC3AA211CBD921

View file

@ -3,47 +3,36 @@
import tempfile import tempfile
import datetime import datetime
import logging
import os.path import os.path
import json import json
from google.auth.transport.requests import Request from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials #from google.oauth2.credentials import Credentials
from google.oauth2.service_account import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build from googleapiclient.discovery import build
from googleapiclient.errors import HttpError from googleapiclient.errors import HttpError
logger = logging.getLogger(f"wp_cal.{__name__}")
# If modifying these scopes, delete the file token.json. # If modifying these scopes, delete the file token.json.
SCOPES = ['https://www.googleapis.com/auth/calendar'] SCOPES = ['https://www.googleapis.com/auth/calendar.readonly']
class Google(): class Google():
def __init__(self, calendar_id, *, credentials, token_file): def __init__(self, calendar_id, *, credentials, token_file):
_, credentials_file = tempfile.mkstemp()
with open(credentials_file, 'w') as fp:
json.dump(credentials, fp)
self.calendar_id = calendar_id self.calendar_id = calendar_id
self.credentials_file = credentials_file self.credentials_info = credentials
self.token_file = token_file
self.credentials = None self.credentials = None
def login(self): def login(self):
if os.path.exists(self.token_file): self.credentials = Credentials.from_service_account_info(self.credentials_info, scopes=SCOPES)
self.credentials = Credentials.from_authorized_user_file(self.token_file, SCOPES) if not self.credentials.valid:
if self.credentials is None or not self.credentials.valid: self.credentials.refresh(Request())
if self.credentials is not None \ logger.debug('credentials.valid = %s', repr(self.credentials.valid))
and self.credentials.expired \
and self.credentials.refresh_token:
self.credentials.refresh(Request())
else:
flow = InstalledAppFlow.from_client_secrets_file(
self.credentials_file,
SCOPES,
)
self.credentials = flow.run_local_server(port=0)
with open(self.token_file, 'w') as token:
token.write(self.credentials.to_json())
def get_events(self, start: datetime.datetime | None=None, until: datetime.timedelta | None=None, limit=None): def get_events(self, start: datetime.datetime | None=None, until: datetime.timedelta | None=None, limit=None):
until = until if until else datetime.timedelta(days=365) until = until if until else datetime.timedelta(days=365)
@ -59,7 +48,7 @@ class Google():
maxResults=limit, maxResults=limit,
).execute() ).execute()
except HttpError: except HttpError:
print('an error occured') logger.exception("a http error occured")
raise raise
events = events.get('items', []) events = events.get('items', [])
return events return events