import logging import google.auth.transport.requests from google.oauth2.credentials import Credentials from googleapiclient.discovery import build from shared.secrets import get_gmail_oauth, save_gmail_tokens logger = logging.getLogger(__name__) _service = None SCOPES = [ "https://www.googleapis.com/auth/gmail.readonly", "https://www.googleapis.com/auth/calendar", ] def get_calendar_service(): global _service oauth = get_gmail_oauth() creds = Credentials( token=oauth.get("access_token"), refresh_token=oauth["refresh_token"], token_uri="https://oauth2.googleapis.com/token", client_id=oauth["client_id"], client_secret=oauth["client_secret"], scopes=SCOPES, ) if not creds.valid: creds.refresh(google.auth.transport.requests.Request()) save_gmail_tokens( creds.token, creds.expiry.isoformat() if creds.expiry else "", ) _service = build("calendar", "v3", credentials=creds, cache_discovery=False) return _service def list_events(service, time_min, time_max, max_results=10): result = ( service.events() .list( calendarId="primary", timeMin=time_min, timeMax=time_max, maxResults=max_results, singleEvents=True, orderBy="startTime", ) .execute() ) return result.get("items", []) def create_event(service, summary, start, end, attendees=None, description=None): body = { "summary": summary, "start": start, "end": end, } if description: body["description"] = description if attendees: body["attendees"] = [{"email": a} for a in attendees] return service.events().insert(calendarId="primary", body=body).execute() def check_availability(service, time_min, time_max): body = { "timeMin": time_min, "timeMax": time_max, "items": [{"id": "primary"}], } result = service.freebusy().query(body=body).execute() return result["calendars"]["primary"]["busy"]