"""Update an event in Google Calendar."""
import re
from datetime import datetime
from typing import Any, Dict, List, Optional, Type, Union
from uuid import uuid4
from zoneinfo import ZoneInfo
from langchain_core.callbacks import CallbackManagerForToolRun
from pydantic import BaseModel, Field
from langchain_google_community.calendar.base import CalendarBaseTool
from langchain_google_community.calendar.utils import is_all_day_event
[docs]
class UpdateEventSchema(BaseModel):
"""Input for CalendarUpdateEvent."""
event_id: str = Field(..., description="The event ID to update.")
calendar_id: str = Field(
default="primary", description="The calendar ID to create the event in."
)
summary: Optional[str] = Field(default=None, description="The title of the event.")
start_datetime: Optional[str] = Field(
default=None,
description=(
"The new start datetime for the event in 'YYYY-MM-DD HH:MM:SS' format. "
"If the event is an all-day event, set the time to 'YYYY-MM-DD' format."
),
)
end_datetime: Optional[str] = Field(
default=None,
description=(
"The new end datetime for the event in 'YYYY-MM-DD HH:MM:SS' format. "
"If the event is an all-day event, set the time to 'YYYY-MM-DD' format."
),
)
timezone: Optional[str] = Field(
default=None, description="The timezone of the event."
)
recurrence: Optional[Dict[str, Any]] = Field(
default=None,
description=(
"The recurrence of the event. "
"Format: {'FREQ': <'DAILY' or 'WEEKLY'>, 'INTERVAL': <number>, "
"'COUNT': <number or None>, 'UNTIL': <'YYYYMMDD' or None>, "
"'BYDAY': <'MO', 'TU', 'WE', 'TH', 'FR', 'SA', 'SU' or None>}. "
"Use either COUNT or UNTIL, but not both; set the other to None."
),
)
location: Optional[str] = Field(
default=None, description="The location of the event."
)
description: Optional[str] = Field(
default=None, description="The description of the event."
)
attendees: Optional[List[str]] = Field(
default=None, description="A list of attendees' email addresses for the event."
)
reminders: Union[None, bool, List[Dict[str, Any]]] = Field(
default=None,
description=(
"Reminders for the event. "
"Set to True for default reminders, or provide a list like "
"[{'method': 'email', 'minutes': <minutes>}, ...]. "
"Valid methods are 'email' and 'popup'."
),
)
conference_data: Optional[bool] = Field(
default=None, description="Whether to include conference data."
)
color_id: Optional[str] = Field(
default=None,
description=(
"The color ID of the event. None for default. "
"'1': Lavender, '2': Sage, '3': Grape, '4': Flamingo, '5': Banana, "
"'6': Tangerine, '7': Peacock, '8': Graphite, '9': Blueberry, "
"'10': Basil, '11': Tomato."
),
)
transparency: Optional[str] = Field(
default=None,
description=(
"User availability for the event."
"transparent for available and opaque for busy."
),
)
send_updates: Optional[str] = Field(
default=None,
description=(
"Whether to send updates to attendees. "
"Allowed values are 'all', 'externalOnly', or 'none'."
),
)
[docs]
class CalendarUpdateEvent(CalendarBaseTool): # type: ignore[override, override]
"""Tool that updates an event in Google Calendar."""
name: str = "update_calendar_event"
description: str = "Use this tool to update an event. "
args_schema: Type[UpdateEventSchema] = UpdateEventSchema
def _get_event(self, event_id: str, calendar_id: str = "primary") -> Dict[str, Any]:
"""Get the event by ID."""
event = (
self.api_resource.events()
.get(calendarId=calendar_id, eventId=event_id)
.execute()
)
return event
def _refactor_event(
self,
event: Dict[str, Any],
summary: Optional[str] = None,
start_datetime: Optional[str] = None,
end_datetime: Optional[str] = None,
timezone: Optional[str] = None,
recurrence: Optional[Dict[str, Any]] = None,
location: Optional[str] = None,
description: Optional[str] = None,
attendees: Optional[List[str]] = None,
reminders: Union[None, bool, List[Dict[str, Any]]] = None,
conference_data: Optional[bool] = None,
color_id: Optional[str] = None,
transparency: Optional[str] = None,
) -> Dict[str, Any]:
"""Refactor the event body."""
if summary is not None:
event["summary"] = summary
try:
if start_datetime and end_datetime:
if is_all_day_event(start_datetime, end_datetime):
event["start"] = {"date": start_datetime}
event["end"] = {"date": end_datetime}
else:
datetime_format = "%Y-%m-%d %H:%M:%S"
timezone = timezone or event["start"]["timeZone"]
start_dt = datetime.strptime(start_datetime, datetime_format)
end_dt = datetime.strptime(end_datetime, datetime_format)
event["start"] = {
"dateTime": start_dt.replace(
tzinfo=ZoneInfo(timezone)
).isoformat(),
"timeZone": timezone,
}
event["end"] = {
"dateTime": end_dt.replace(
tzinfo=ZoneInfo(timezone)
).isoformat(),
"timeZone": timezone,
}
except ValueError as error:
raise ValueError("The datetime format is incorrect.") from error
if (recurrence is not None) and (isinstance(recurrence, dict)):
recurrence_items = [
f"{k}={v}" for k, v in recurrence.items() if v is not None
]
event.update({"recurrence": ["RRULE:" + ";".join(recurrence_items)]})
if location is not None:
event.update({"location": location})
if description is not None:
event.update({"description": description})
if attendees is not None:
attendees_emails = []
email_pattern = r"^[^@]+@[^@]+\.[^@]+$"
for email in attendees:
if not re.match(email_pattern, email):
raise ValueError(f"Invalid email address: {email}")
attendees_emails.append({"email": email})
event.update({"attendees": attendees_emails})
if reminders is not None:
if reminders is True:
event.update({"reminders": {"useDefault": True}})
elif isinstance(reminders, list):
for reminder in reminders:
if "method" not in reminder or "minutes" not in reminder:
raise ValueError(
"Each reminder must have 'method' and 'minutes' keys."
)
if reminder["method"] not in ["email", "popup"]:
raise ValueError(
"The reminder method must be 'email' or 'popup'."
)
event.update(
{"reminders": {"useDefault": False, "overrides": reminders}}
)
else:
event.update({"reminders": {"useDefault": False}})
if conference_data:
event.update(
{
"conferenceData": {
"createRequest": {
"requestId": str(uuid4()),
"conferenceSolutionKey": {"type": "hangoutsMeet"},
}
}
}
)
else:
event.update({"conferenceData": None})
if color_id is not None:
event["colorId"] = color_id
if transparency is not None:
event.update({"transparency": transparency})
return event
def _run(
self,
event_id: str,
summary: str,
start_datetime: str,
end_datetime: str,
calendar_id: str = "primary",
timezone: Optional[str] = None,
recurrence: Optional[Dict[str, Any]] = None,
location: Optional[str] = None,
description: Optional[str] = None,
attendees: Optional[List[str]] = None,
reminders: Union[None, bool, List[Dict[str, Any]]] = None,
conference_data: Optional[bool] = None,
color_id: Optional[str] = None,
transparency: Optional[str] = None,
send_updates: Optional[str] = None,
run_manager: Optional[CallbackManagerForToolRun] = None,
) -> str:
"""Run the tool to update an event in Google Calendar."""
try:
event = self._get_event(event_id, calendar_id)
body = self._refactor_event(
event=event,
summary=summary,
start_datetime=start_datetime,
end_datetime=end_datetime,
timezone=timezone,
recurrence=recurrence,
location=location,
description=description,
attendees=attendees,
reminders=reminders,
conference_data=conference_data,
color_id=color_id,
transparency=transparency,
)
conference_version = 1 if conference_data else 0
result = (
self.api_resource.events()
.update(
calendarId=calendar_id,
eventId=event_id,
body=body,
conferenceDataVersion=conference_version,
sendUpdates=send_updates,
)
.execute()
)
return f"Event updated: {result.get('htmlLink')}"
except Exception as error:
raise Exception(f"An error occurred: {error}") from error
