microsoft-365-service/app/schemas.py

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")