Source code for langchain_google_community.calendar.update_event

"""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 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.astimezone().isoformat(),
                        "timeZone": timezone,
                    }
                    event["end"] = {
                        "dateTime": end_dt.astimezone().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