Graph Mail implemented

ITK_dev_shared_components/graph/authentication.py

@@ -0,0 +1,65 @@
+"""This module is responsible for authenticating a Microsoft Graph
+connection."""
+
+import msal
+
+def authorize_by_username_password(username: str, password: str, *, client_id: str, tenant_id: str) -> msal.PublicClientApplication:
+    """Get a bearer token for the given user.
+    This is used in most other Graph API calls.
+
+    Args:
+        username: The username of the user (email address).
+        password: The password of the user.
+        client_id: The Graph API client id in 8-4-4-12 format.
+        tenant_id: The Graph API tenant id in 8-4-4-12 format.
+
+    Returns:
+        str: The bearer token.
+
+    Raises:
+        RuntimeError: If the token couldn't be acquired.
+    """
+    authority = f"https://login.microsoftonline.com/{tenant_id}"
+    scopes = ["https://graph.microsoft.com/.default"]
+
+    app = msal.PublicClientApplication(client_id, authority=authority)
+    app.acquire_token_by_username_password(username, password, scopes)
+
+    graph_access = GraphAccess(app, scopes)
+
+    # Test connection
+    graph_access.get_access_token()
+
+    return graph_access
+
+
+class GraphAccess:
+    """An object that handles access to the Graph api.
+    This object should not be created directly but instead
+    using one of the authorize methods in the graph.authentication module.
+    """
+    def __init__(self, app: msal.PublicClientApplication, scopes: list[str]) -> str:
+        self.app = app
+        self.scopes = scopes
+
+    def get_access_token(self):
+        """Get the access token to Graph.
+        This function automatically reuses an existing token 
+        or refreshes an expired one.
+
+        Raises:
+            RuntimeError: If the access token couldn't be acquired.
+
+        Returns:
+            str: The Graph access token.
+        """
+        account = self.app.get_accounts()[0]
+        token = self.app.acquire_token_silent(self.scopes, account)
+
+        if "access_token" in token:
+            return token['access_token']
+
+        if 'error_description' in token:
+            raise RuntimeError(f"Token could not be acquired. {token['error_description']}")
+
+        raise RuntimeError("Something went wrong. No error description was returned from Graph.")

ITK_dev_shared_components/graph/mail.py

@@ -0,0 +1,332 @@
+"""This module is responsible for accessing emails using the Microsoft Graph API."""
+
+from dataclasses import dataclass, field
+import io
+
+import requests
+from bs4 import BeautifulSoup
+
+from ITK_dev_shared_components.graph.authentication import GraphAccess
+
+
+@dataclass
+class Email:
+    """A class representing an email."""
+    user: str
+    id: str = field(repr=False)
+    received_time: str
+    sender: str
+    receivers: list[str]
+    subject: str
+    body: str = field(repr=False)
+    body_type: str
+    has_attachments: bool
+
+    def get_text(self) -> str:
+        """Get the body as plain text.
+        If the body is html it's converted to plaintext.
+        If the body is text it's returned as is.
+
+        Returns:
+            str: The body as plain text.
+        """
+        if self.body_type == 'html':
+            soup = BeautifulSoup(self.body, "html.parser")
+            return soup.get_text().strip()
+
+        return self.body
+
+
+@dataclass
+class Attachment:
+    """A dataclass representing an email Attachment.
+    It contains the graph id, name and size of the attachment.
+    To get the actual data call graph.mail.get_attachment_data.
+    """
+    email: Email = field(repr=False)
+    id: str = field(repr=False)
+    name: str
+    size: int
+
+
+def get_emails_from_folder(user: str, folder_path: str, graph_access: GraphAccess) -> tuple[Email]:
+    """Get all emails from the specified user and folder.
+    You need to authorize against Graph to get the GraphAccess before using this function
+    see the graph.authentication module.
+
+    Args:
+        user: The user who owns the folder.
+        folder_path: The absolute path of the folder e.g. 'Inbox/Economy/May'
+        graph_access: The GraphAccess object used to authenticate.
+
+    Returns:
+        tuple[Email]: The emails from the given folder.
+    """
+    folder_id = get_folder_id_from_path(user, folder_path, graph_access)
+
+    endpoint = f"https://graph.microsoft.com/v1.0/users/{user}/mailFolders/{folder_id}/messages?$top=1000"
+
+    response = _get_request(endpoint, graph_access)
+    emails_raw = response.json()['value']
+
+    return _unpack_email_response(user, emails_raw)
+
+
+def get_email_as_mime(email: Email, graph_access: GraphAccess) -> io.BytesIO:
+    """Get an email as a file-like object in MIME format.
+
+    Args:
+        email: The email to get as MIME.
+        graph_access: The GraphAccess object used to authenticate.
+
+    Returns:
+        io.BytesIO: A file-like object of the MIME file.
+    """
+    endpoint = f"https://graph.microsoft.com/v1.0/users/{email.user}/messages/{email.id}/$value"
+    response = _get_request(endpoint, graph_access)
+    data = response.content
+    return io.BytesIO(data)
+
+
+def get_folder_id_from_path(user: str, folder_path: str, graph_access: GraphAccess) -> str:
+    """Get the Graph id of a folder based on the path of the folder.
+    You need to authorize against Graph to get the GraphAccess before using this function
+    see the graph.authentication module.
+
+    Args:
+        user: The user who owns the folder.
+        folder_path: The absolute path of the folder e.g. 'Inbox/Economy/May'
+        graph_access: The GraphAccess object used to authenticate.
+
+    Raises:
+        ValueError: If a folder in the path can't be found.
+
+    Returns:
+        str: The UUID of the folder in Graph.
+    """
+    folders = folder_path.split("/")
+    main_folder = folders[0]
+    child_folders = folders[1:]
+
+    folder_id = None
+
+    # Get main folder
+    endpoint = f"https://graph.microsoft.com/v1.0/users/{user}/mailFolders"
+    response = _get_request(endpoint, graph_access).json()
+    folder_id = _find_folder(response, main_folder)
+    if folder_id is None:
+        raise ValueError(f"Top level folder '{main_folder}' was not found for user '{user}'.")
+
+    # Get child folders
+    for child_folder in child_folders:
+        endpoint = f"https://graph.microsoft.com/v1.0/users/{user}/mailFolders/{folder_id}/childFolders"
+        response = _get_request(endpoint, graph_access).json()
+        folder_id = _find_folder(response, child_folder)
+        if folder_id is None:
+            raise ValueError(f"Child folder '{child_folder}' not found under '{main_folder}' for user '{user}'.")
+
+    return folder_id
+
+
+def list_email_attachments(email: Email, graph_access: GraphAccess) -> tuple[Attachment]:
+    """List all attachments of the given email. This function only gets the id, name and size
+    of the attachment. Use get_attachment_data to get the actual data of an attachment.
+
+    Args:
+        email: The email which attachments to list.
+        graph_access: The GraphAccess object used to authenticate.
+
+    Returns:
+        tuple[Attachment]: A tuple of Attachment objects describing the attachments.
+    """
+    endpoint = f"https://graph.microsoft.com/v1.0/users/{email.user}/messages/{email.id}/attachments?$select=name,size,id"
+    response = _get_request(endpoint, graph_access).json()
+
+    attachments = []
+    for att in response['value']:
+        attachments.append(Attachment(email, att['id'], att['name'], att['size']))
+
+    return tuple(attachments)
+
+
+def get_attachment_data(attachment: Attachment, graph_access: GraphAccess) -> io.BytesIO:
+    """Get a file-like object representing the attachment.
+
+    Args:
+        attachment: The attachment to get.
+        graph_access: The GraphAccess object used to authenticate.
+
+    Returns:
+        io.BytesIO: A file-like object representing the attachment.
+    """
+    email = attachment.email
+    endpoint = f"https://graph.microsoft.com/v1.0/users/{email.user}/messages/{email.id}/attachments/{attachment.id}/$value"
+    response =  _get_request(endpoint, graph_access)
+    data_bytes = response.content
+    return io.BytesIO(data_bytes)
+
+
+def move_email(email: Email, folder_path: str, graph_access: GraphAccess, *, well_known_folder: bool=False) -> None:
+    """Move an email to another folder under the same user.
+    If well_known_folder is true, the folder path is assumed to be a well defined folder.
+    See https://learn.microsoft.com/en-us/graph/api/resources/mailfolder?view=graph-rest-1.0 
+    for a list of well defined folder names.
+
+    Args:
+        email: The email to move.
+        folder_path: The absolute path to the new folder. E.g. 'Inbox/Economy/May'
+        graph_access: The GraphAccess object used to authenticate.
+        well_known_folder: Whether the path is a 'well known folder'. Defaults to False.
+    """
+    if well_known_folder:
+        folder_id = folder_path
+    else:
+        folder_id = get_folder_id_from_path(email.user, folder_path, graph_access)
+
+    endpoint = f"https://graph.microsoft.com/v1.0/users/{email.user}/messages/{email.id}/move"
+
+    token = graph_access.get_access_token()
+    headers = {
+        'Authorization': f"Bearer {token}",
+        'Content-Type': "application/json"
+    }
+
+    body = {
+        'destinationId': folder_id
+    }
+
+    response = requests.post(
+        url=endpoint,
+        headers=headers,
+        json=body,
+        timeout=30
+    )
+    response.raise_for_status()
+
+    new_id = response.json()['id']
+    email.id = new_id
+
+
+def delete_email(email: Email, graph_access: GraphAccess, *, permanent: bool=False) -> None:
+    """Delete an email from the mailbox.
+    If permanent is true the email is completely removed from the user's mailbox.
+    If permanent is false the email is instead moved to the Deleted Items folder.
+
+    Args:
+        email: The email to delete.
+        graph_access: The GraphAccess object used to authenticate.
+        permanent: Whether to permanently remove the email or not. Defaults to False.
+    """
+    if permanent:
+        endpoint = f"https://graph.microsoft.com/v1.0/users/{email.user}/messages/{email.id}"
+
+        token = graph_access.get_access_token()
+        headers = {'Authorization': f"Bearer {token}"}
+
+        response = requests.delete(
+            url=endpoint,
+            headers=headers,
+            timeout=30
+        )
+        response.raise_for_status()
+    else:
+        move_email(email, "deleteditems", graph_access, well_known_folder=True)
+
+
+
+
+def _find_folder(response: dict, target_folder: str) -> str:
+    for g_folder in response['value']:
+        if g_folder['displayName'] == target_folder:
+            return g_folder['id']
+    return None
+
+
+def _unpack_email_response(user: str, emails_raw: list[dict[str, str]]) -> tuple[Email]:
+    """Unpack a json HTTP response and create a list of Email objects.
+
+    Args:
+        user: The user who owns the email folder.
+        json: The json dictionary created by response.json().
+
+    Returns:
+        tuple[Email]: A tuple of Email objects.
+    """
+    emails = []
+
+    for email in emails_raw:
+        mail_id = email['id']
+        received_time = email['receivedDateTime']
+        sender = email['from']['emailAddress']['address']
+        receivers = [r['emailAddress']['address'] for r in email['toRecipients']]
+        subject = email['subject']
+        body = email['body']['content']
+        body_type = email['body']['contentType']
+        has_attachments = email['hasAttachments']
+
+        emails.append(
+            Email(
+                user,
+                mail_id,
+                received_time,
+                sender,
+                receivers,
+                subject,
+                body,
+                body_type,
+                has_attachments
+            )
+        )
+
+    return tuple(emails)
+
+
+def _get_request(endpoint: str, graph_access: GraphAccess) -> requests.models.Response:
+    """Sends a get request to the given Graph endpoint using the GraphAccess
+    and returns the json object of the response.
+
+    Args:
+        endpoint: The URL of the Graph endpoint.
+        graph_access: The GraphAccess object used to authenticate.
+        timeout: Timeout in seconds of the HTTP request. Defaults to 10.
+
+    Returns:
+        Response: The response object of the GET request.
+
+    Raises:
+        HTTPError: Any errors raised while performing GET request.
+    """
+    token = graph_access.get_access_token()
+    headers = {'Authorization': f"Bearer {token}"}
+
+    response = requests.get(
+        endpoint,
+        headers=headers,
+        timeout=30
+    )
+    response.raise_for_status()
+
+    return response
+
+
+if __name__ == '__main__':
+    import os, json, authentication
+    def main():
+        credentials = json.loads(os.environ['GraphAPI'])
+        client_id = credentials['client_id']
+        tenant_id = credentials['tenant_id']
+        username = credentials['username']
+        password = credentials['password']
+
+        graph_access = authentication.authorize_by_username_password(username, password, client_id=client_id, tenant_id=tenant_id)
+
+        emails = get_emails_from_folder("itk-rpa@mkb.aarhus.dk", "Indbakke/Graph Test/Undermappe", graph_access)
+        email = emails[0]
+
+        print(repr(email.get_text()))
+
+        # move_email(email, "Indbakke/Graph Test/Undermappe2", graph_access)
+
+        # list_email_attachments(email, graph_access)
+
+    main()