from typing import Any from pydantic import BaseModel, EmailStr, Field class EmailAddress(BaseModel): address: EmailStr name: str | None = None def as_graph_recipient(self) -> dict[str, Any]: payload: dict[str, Any] = {"address": str(self.address)} if self.name: payload["name"] = self.name return {"emailAddress": payload} class FileAttachment(BaseModel): name: str content_type: str = "application/octet-stream" content_bytes_base64: str = Field(..., description="Base64 encoded file content.") def as_graph_attachment(self) -> dict[str, Any]: return { "@odata.type": "#microsoft.graph.fileAttachment", "name": self.name, "contentType": self.content_type, "contentBytes": self.content_bytes_base64, } class SendMailRequest(BaseModel): subject: str body: str to: list[EmailAddress] cc: list[EmailAddress] = Field(default_factory=list) bcc: list[EmailAddress] = Field(default_factory=list) reply_to: list[EmailAddress] = Field(default_factory=list) body_content_type: str = Field("HTML", pattern="^(HTML|Text)$") save_to_sent_items: bool = True attachments: list[FileAttachment] = Field(default_factory=list) class CalendarAttendee(BaseModel): email: EmailStr name: str | None = None type: str = Field("required", pattern="^(required|optional|resource)$") def as_graph_attendee(self) -> dict[str, Any]: email_address: dict[str, Any] = {"address": str(self.email)} if self.name: email_address["name"] = self.name return {"emailAddress": email_address, "type": self.type} class DateTimeTimeZone(BaseModel): date_time: str = Field(..., alias="dateTime") time_zone: str = Field("UTC", alias="timeZone") class CalendarEventRequest(BaseModel): subject: str body: str | None = None body_content_type: str = Field("HTML", pattern="^(HTML|Text)$") start: DateTimeTimeZone end: DateTimeTimeZone location: str | None = None attendees: list[CalendarAttendee] = Field(default_factory=list) is_online_meeting: bool = False online_meeting_provider: str | None = Field(None, pattern="^(teamsForBusiness|skypeForBusiness|skypeForConsumer)$") class CalendarEventUpdateRequest(BaseModel): """Partial update; only fields that are set are sent to Microsoft Graph.""" subject: str | None = None body: str | None = None body_content_type: str = Field("HTML", pattern="^(HTML|Text)$") start: DateTimeTimeZone | None = None end: DateTimeTimeZone | None = None location: str | None = None attendees: list[CalendarAttendee] | None = None is_online_meeting: bool | None = None online_meeting_provider: str | None = Field(None, pattern="^(teamsForBusiness|skypeForBusiness|skypeForConsumer)$") def as_graph_patch(self) -> dict[str, Any]: payload: dict[str, Any] = {} if self.subject is not None: payload["subject"] = self.subject if self.body is not None: payload["body"] = {"contentType": self.body_content_type, "content": self.body} if self.start is not None: payload["start"] = self.start.model_dump(by_alias=True) if self.end is not None: payload["end"] = self.end.model_dump(by_alias=True) if self.location is not None: payload["location"] = {"displayName": self.location} if self.attendees is not None: payload["attendees"] = [attendee.as_graph_attendee() for attendee in self.attendees] if self.is_online_meeting is not None: payload["isOnlineMeeting"] = self.is_online_meeting if self.online_meeting_provider is not None: payload["onlineMeetingProvider"] = self.online_meeting_provider return payload class ScheduleRequest(BaseModel): """Free/busy lookup (Graph getSchedule) for one or more mailboxes.""" schedules: list[EmailStr] = Field(..., min_length=1, description="Mailboxes whose availability is requested.") start: DateTimeTimeZone end: DateTimeTimeZone availability_view_interval: int = Field(30, ge=5, le=1440, description="Slot length in minutes.") def as_graph_payload(self) -> dict[str, Any]: return { "schedules": [str(address) for address in self.schedules], "startTime": self.start.model_dump(by_alias=True), "endTime": self.end.model_dump(by_alias=True), "availabilityViewInterval": self.availability_view_interval, } def _planner_assignments(assign_user_ids: list[str], unassign_user_ids: list[str]) -> dict[str, Any]: assignments: dict[str, Any] = { user_id: {"@odata.type": "#microsoft.graph.plannerAssignment", "orderHint": " !"} for user_id in assign_user_ids } for user_id in unassign_user_ids: assignments[user_id] = None return assignments class PlannerTaskCreateRequest(BaseModel): plan_id: str bucket_id: str | None = None title: str start_date_time: str | None = Field(None, description="ISO 8601 with offset, e.g. 2026-09-08T08:00:00Z.") due_date_time: str | None = Field(None, description="ISO 8601 with offset, e.g. 2026-09-10T16:00:00Z.") percent_complete: int | None = Field(None, ge=0, le=100) priority: int | None = Field(None, ge=0, le=10, description="Graph priority: 1 urgent, 3 important, 5 medium, 9 low.") assign_user_ids: list[str] = Field(default_factory=list, description="Microsoft Graph user ids to assign.") def as_graph_payload(self) -> dict[str, Any]: payload: dict[str, Any] = {"planId": self.plan_id, "title": self.title} if self.bucket_id: payload["bucketId"] = self.bucket_id if self.start_date_time: payload["startDateTime"] = self.start_date_time if self.due_date_time: payload["dueDateTime"] = self.due_date_time if self.percent_complete is not None: payload["percentComplete"] = self.percent_complete if self.priority is not None: payload["priority"] = self.priority if self.assign_user_ids: payload["assignments"] = _planner_assignments(self.assign_user_ids, []) return payload class PlannerTaskUpdateRequest(BaseModel): """Partial update; only fields that are set are sent to Microsoft Graph.""" title: str | None = None bucket_id: str | None = None start_date_time: str | None = None due_date_time: str | None = None percent_complete: int | None = Field(None, ge=0, le=100) priority: int | None = Field(None, ge=0, le=10) assign_user_ids: list[str] = Field(default_factory=list) unassign_user_ids: list[str] = Field(default_factory=list) def as_graph_patch(self) -> dict[str, Any]: payload: dict[str, Any] = {} if self.title is not None: payload["title"] = self.title if self.bucket_id is not None: payload["bucketId"] = self.bucket_id if self.start_date_time is not None: payload["startDateTime"] = self.start_date_time if self.due_date_time is not None: payload["dueDateTime"] = self.due_date_time if self.percent_complete is not None: payload["percentComplete"] = self.percent_complete if self.priority is not None: payload["priority"] = self.priority if self.assign_user_ids or self.unassign_user_ids: payload["assignments"] = _planner_assignments(self.assign_user_ids, self.unassign_user_ids) return payload class DriveUploadRequest(BaseModel): content_base64: str content_type: str = "application/octet-stream" class GraphCollectionResponse(BaseModel): value: list[dict[str, Any]] next_link: str | None = Field(None, alias="@odata.nextLink")