Google Calendar
Keep each of your users' meetings in their memory, so their assistant knows who they are meeting, when and
about what, and can prepare them for it. Your app already holds an access token for each user, from their
Google sign-in with the calendar.readonly scope; this sync adds each event on their main calendar under
its own id with the label channel: calendar. The first run reads the whole calendar. After that,
Google's sync token says what changed: a moved meeting teaches its new time, and a cancelled one is deleted
from memory with what it taught.
Install
pip install httpx geniffyuv add httpx geniffySet GENIFFY_API_KEY from API keys in the Geniffy app.
The sync
import httpx
from geniffy import Geniffy, NotFoundError
geniffy = Geniffy() # reads GENIFFY_API_KEY
LABELS = {"channel": "calendar"}
def calendar(token: str) -> httpx.Client:
"""One user's calendar, with the access token from their Google sign-in."""
return httpx.Client(base_url="https://www.googleapis.com/calendar/v3/calendars/primary", timeout=30,
headers={"Authorization": f"Bearer {token}"})
def note(event: dict) -> str:
"""An event as one note: what it is, when, with whom, where, and what it is about."""
start, end = event["start"], event["end"]
when = (f"{start['date']}, all day" if "date" in start
else f"{start['dateTime']} to {end['dateTime']}")
people = [p.get("displayName") or p["email"] for p in event.get("attendees", []) if not p.get("resource")]
lines = [f"Meeting: {event.get('summary') or '(no title)'}", f"When: {when}"]
if people:
lines.append("With: " + ", ".join(people))
if event.get("location"):
lines.append("Where: " + event["location"])
if event.get("description"):
lines.append(event["description"])
return "\n".join(lines)
def sync(user_id: str, token: str, sync_token: str | None = None) -> str:
"""Bring one user's calendar into their memory. Returns the token to pass next time: pass None the first
time and the whole calendar is read; after that, only the events that changed."""
mem, seen = geniffy.space(f"user_{user_id}"), set()
params = {"syncToken": sync_token} if sync_token else {}
with calendar(token) as api:
while True:
got = api.get("/events", params={**params, "maxResults": 250})
if got.status_code == 410: # the token has run out: read everything again
return sync(user_id, token)
out = got.raise_for_status().json()
for event in out.get("items", []):
ref = f"calendar:{event['id']}"
if event["status"] == "cancelled":
try:
mem.sources.delete(external_id=ref)
except NotFoundError:
pass
continue
mem.memories.add(note(event), title=event.get("summary") or "(no title)", external_id=ref,
labels=LABELS)
seen.add(ref)
if "nextPageToken" not in out:
break
params["pageToken"] = out["nextPageToken"]
if sync_token is None: # everything was read: what is no longer there goes
mem.sources.delete_labelled(LABELS, keep=seen)
return out["nextSyncToken"]
def disconnect(user_id: str) -> int:
"""The user disconnected their calendar: everything that came from it goes. Returns how many events."""
return geniffy.space(f"user_{user_id}").sources.delete_labelled(LABELS)Run it on a schedule, and keep the token each run returns with the user for the next run. Google's access tokens last an hour, so refresh the user's before each run, as your Google sign-in library does.
How it behaves
- Each event is one source, under its Google id. A meeting moved to another day sends the event again, and memory learns the new time and lets the old one go. See Your own ids.
- Only what changed is read. After the first run, the sync token names the events that changed; when Google lets it expire, the run reads everything again and puts memory right.
- A cancelled event goes from memory, with what it taught.
- People, not rooms. Meeting rooms booked as attendees are left out of who the meeting is with.
- Recall can keep to the calendar:
mem.context(question, labels={"channel": "calendar"}). See Labels. - Disconnecting forgets it all in one call, and nothing the user added another way.