from datetime import datetime import os.path from base64 import b16encode from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow from googleapiclient.discovery import build, Resource from googleapiclient.errors import HttpError # If modifying these scopes, delete the file token.json. SCOPES = [ "https://www.googleapis.com/auth/calendar.readonly", "https://www.googleapis.com/auth/calendar.events", ] class CalendarService: service: Resource def __init__(self): creds = self._authenticate() self.service = build("calendar", "v3", credentials=creds) @staticmethod def encode_event_id(id): return b16encode(id.encode("ascii")).decode("ascii").lower().rstrip("=") def _authenticate(self) -> Credentials: creds = None # The file token.json stores the user's access and refresh tokens, and is # created automatically when the authorization flow completes for the first # time. if os.path.exists("token.json"): creds = Credentials.from_authorized_user_file("token.json", SCOPES) # If there are no (valid) credentials available, let the user log in. if not creds or not creds.valid: if creds and creds.expired and creds.refresh_token: creds.refresh(Request()) else: flow = InstalledAppFlow.from_client_secrets_file( "credentials.json", SCOPES ) creds = flow.run_local_server(port=0) # Save the credentials for the next run with open("token.json", "w") as token: token.write(creds.to_json()) return creds def insert_or_update_event( self, calendar_id: str, id: str, title: str, start: datetime, end: datetime ): event_id = self.encode_event_id(id) event = { "id": event_id, "summary": title, "start": { "dateTime": start.isoformat() + "Z", # 'Z' indicates UTC time }, "end": { "dateTime": end.isoformat() + "Z", # 'Z' indicates UTC time }, } print(f"Adding/updating event: {title}, start: {start}, end: {end}") try: self.service.events().update( calendarId=calendar_id, eventId=event_id, body=event ).execute() except HttpError as error: # Create event if it doesn't exist if error.status_code == 404: self.service.events().insert( calendarId=calendar_id, body=event ).execute() else: raise def delete_event(self, calendar_id: str, id: str): event_id = self.encode_event_id(id) try: self.service.events().delete(calendar_id, event_id) except HttpError as error: print("Got error when deleting event: ", error)