Fixed Bug in graph.py (#398)

* Fixed Bug in graph.py

Fixed Bug regarding the finding of a specific folder.
This Bug caused parsedmarc to crash if it could not find the folder in one Ms Graph request. This is only an issue if your MailBox contains 10+ folders.
It was solved by adding the `$filter=displayName eq '{folder_name}'` param so it would immediatly find the folder.

* Fixed MS Graph Search bug

Fixed bug that causes only 10 messages to be read from MS Graph if batch size is defined as 0.
Fixed formatting

* prevented $top from being 0
This commit is contained in:
rubeste
2023-05-03 21:36:23 +02:00
committed by GitHub
parent 9c86e2df49
commit 70a9d11adb

View File

@@ -145,17 +145,34 @@ class MSGraphConnection(MailboxConnection):
folder_id = self._find_folder_id_from_folder_path(folder_name)
url = f'/users/{self.mailbox_name}/mailFolders/' \
f'{folder_id}/messages'
batch_size = kwargs.get('batch_size')
if not batch_size:
batch_size = 0
emails = self._get_all_messages(url, batch_size)
return [email['id'] for email in emails]
def _get_all_messages(self, url, batch_size):
messages: list
params = {
'$select': 'id'
}
batch_size = kwargs.get('batch_size')
if batch_size and batch_size > 0:
params['$top'] = batch_size
else:
params['$top'] = 100
result = self._client.get(url, params=params)
if result.status_code != 200:
raise RuntimeError(f'Failed to fetch messages {result.text}')
emails = result.json()['value']
return [email['id'] for email in emails]
messages = result.json()['value']
# Loop if next page is present and not obtained message limit.
while '@odata.nextLink' in result.json() and (
batch_size == 0 or
batch_size - len(messages) > 0):
result = self._client.get(result.json()['@odata.nextLink'])
if result.status_code != 200:
raise RuntimeError(f'Failed to fetch messages {result.text}')
messages.extend(result.json()['value'])
return messages
def mark_message_read(self, message_id: str):
"""Marks a message as read"""
@@ -223,11 +240,12 @@ class MSGraphConnection(MailboxConnection):
if parent_folder_id is not None:
sub_url = f'/{parent_folder_id}/childFolders'
url = f'/users/{self.mailbox_name}/mailFolders{sub_url}'
folders_resp = self._client.get(url)
filter = f"?$filter=displayName eq '{folder_name}'"
folders_resp = self._client.get(url + filter)
if folders_resp.status_code != 200:
raise RuntimeWarning(f"Failed to list folders."
f"{folders_resp.json()}")
folders = folders_resp.json()['value']
folders: list = folders_resp.json()['value']
matched_folders = [folder for folder in folders
if folder['displayName'] == folder_name]
if len(matched_folders) == 0: