microsoft-365-service/app/services.py

import base64
import binascii
from typing import Any
from urllib.parse import quote

from fastapi import HTTPException, status

from .graph_client import MicrosoftGraphClient
from .schemas import CalendarEventRequest, DriveUploadRequest, SendMailRequest


def graph_segment(value: str) -> str:
    return quote(value, safe="")


def graph_path(value: str) -> str:
    return quote(value.strip("/"), safe="/")


class UsersService:
    def __init__(self, graph: MicrosoftGraphClient) -> None:
        self._graph = graph

    async def list_users(self, top: int = 25, search: str | None = None) -> Any:
        params: dict[str, Any] = {"$top": top}
        headers = None
        if search:
            params["$search"] = f'"displayName:{search}" OR "mail:{search}" OR "userPrincipalName:{search}"'
            headers = {"ConsistencyLevel": "eventual"}
        return await self._graph.request("GET", "/users", params=params, headers=headers)

    async def get_user(self, user_id: str) -> Any:
        return await self._graph.request("GET", f"/users/{graph_segment(user_id)}")


class MailService:
    def __init__(self, graph: MicrosoftGraphClient) -> None:
        self._graph = graph

    async def list_messages(self, user_id: str, folder: str = "Inbox", top: int = 25) -> Any:
        params = {
            "$top": top,
            "$orderby": "receivedDateTime desc",
            "$select": "id,subject,from,toRecipients,receivedDateTime,hasAttachments,isRead,webLink",
        }
        return await self._graph.request(
            "GET",
            f"/users/{graph_segment(user_id)}/mailFolders/{graph_segment(folder)}/messages",
            params=params,
        )

    async def send_mail(self, user_id: str, request: SendMailRequest) -> None:
        message: dict[str, Any] = {
            "subject": request.subject,
            "body": {
                "contentType": request.body_content_type,
                "content": request.body,
            },
            "toRecipients": [recipient.as_graph_recipient() for recipient in request.to],
        }
        if request.cc:
            message["ccRecipients"] = [recipient.as_graph_recipient() for recipient in request.cc]
        if request.bcc:
            message["bccRecipients"] = [recipient.as_graph_recipient() for recipient in request.bcc]
        if request.reply_to:
            message["replyTo"] = [recipient.as_graph_recipient() for recipient in request.reply_to]
        if request.attachments:
            message["attachments"] = [attachment.as_graph_attachment() for attachment in request.attachments]

        await self._graph.request(
            "POST",
            f"/users/{graph_segment(user_id)}/sendMail",
            json={"message": message, "saveToSentItems": request.save_to_sent_items},
        )


class CalendarService:
    def __init__(self, graph: MicrosoftGraphClient) -> None:
        self._graph = graph

    async def list_events(self, user_id: str, top: int = 25) -> Any:
        params = {
            "$top": top,
            "$orderby": "start/dateTime",
            "$select": "id,subject,start,end,location,attendees,webLink,isOnlineMeeting,onlineMeeting",
        }
        return await self._graph.request("GET", f"/users/{graph_segment(user_id)}/events", params=params)

    async def create_event(self, user_id: str, request: CalendarEventRequest) -> Any:
        payload: dict[str, Any] = {
            "subject": request.subject,
            "start": request.start.model_dump(by_alias=True),
            "end": request.end.model_dump(by_alias=True),
            "attendees": [attendee.as_graph_attendee() for attendee in request.attendees],
            "isOnlineMeeting": request.is_online_meeting,
        }
        if request.body is not None:
            payload["body"] = {"contentType": request.body_content_type, "content": request.body}
        if request.location:
            payload["location"] = {"displayName": request.location}
        if request.online_meeting_provider:
            payload["onlineMeetingProvider"] = request.online_meeting_provider
        return await self._graph.request("POST", f"/users/{graph_segment(user_id)}/events", json=payload)


class DriveService:
    def __init__(self, graph: MicrosoftGraphClient) -> None:
        self._graph = graph

    async def list_root_children(self, user_id: str, top: int = 25) -> Any:
        return await self._graph.request(
            "GET",
            f"/users/{graph_segment(user_id)}/drive/root/children",
            params={"$top": top},
        )

    async def upload_small_file(self, user_id: str, path: str, request: DriveUploadRequest) -> Any:
        try:
            content = base64.b64decode(request.content_base64, validate=True)
        except binascii.Error as exc:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="content_base64 must contain valid base64 data.",
            ) from exc
        return await self._graph.request(
            "PUT",
            f"/users/{graph_segment(user_id)}/drive/root:/{graph_path(path)}:/content",
            content=content,
            headers={"Content-Type": request.content_type},
        )


class GroupsService:
    def __init__(self, graph: MicrosoftGraphClient) -> None:
        self._graph = graph

    async def list_groups(self, top: int = 25) -> Any:
        params = {"$top": top, "$select": "id,displayName,mail,groupTypes,securityEnabled,mailEnabled"}
        return await self._graph.request("GET", "/groups", params=params)

    async def list_team_channels(self, team_id: str) -> Any:
        return await self._graph.request("GET", f"/teams/{graph_segment(team_id)}/channels")