diff --git a/README.rst b/README.rst index 4124650..a31e35d 100644 --- a/README.rst +++ b/README.rst @@ -45,7 +45,7 @@ Basic subjects = [msg.subject for msg in mailbox.fetch(AND(all=True))] mailbox.logout() -MailBox/MailBoxUnencrypted for create mailbox instance. +MailBox/MailBoxUnencrypted - for create mailbox instance. MailBox.box - imaplib.IMAP4/IMAP4_SSL client instance. @@ -207,7 +207,7 @@ use 'limit' argument for fetch in this case. mailbox.delete([msg.uid for msg in mailbox.fetch()]) # FLAG unseen messages in current folder as Answered and Flagged, *in bulk. - flags = (imap_tools.MessageFlags.ANSWERED, imap_tools.MessageFlags.FLAGGED) + flags = (imap_tools.MailMessageFlags.ANSWERED, imap_tools.MailMessageFlags.FLAGGED) mailbox.flag(mailbox.fetch('(UNSEEN)'), flags, True) # SEEN: mark all messages sent at 05.03.2007 in current folder as unseen, *in bulk @@ -237,6 +237,11 @@ Actions with mailbox folders folder_status = mailbox.folder.status('some_folder') print(folder_status) # {'MESSAGES': 41, 'RECENT': 0, 'UIDNEXT': 11996, 'UIDVALIDITY': 1, 'UNSEEN': 5} +Exceptions +^^^^^^^^^^ + +Custom lib exceptions here: `errors.py `_. + Reasons ------- @@ -271,3 +276,4 @@ Thanks to: * `daitangio `_ * `upils `_ * `Foosec `_ +* `frispete `_ diff --git a/imap_tools/__init__.py b/imap_tools/__init__.py index 1439fd7..44aa39f 100644 --- a/imap_tools/__init__.py +++ b/imap_tools/__init__.py @@ -3,5 +3,6 @@ from .message import * from .folder import * from .utils import * +from .errors import * -__version__ = '0.17.0' +__version__ = '0.18.0' diff --git a/imap_tools/errors.py b/imap_tools/errors.py new file mode 100644 index 0000000..aeb4d1b --- /dev/null +++ b/imap_tools/errors.py @@ -0,0 +1,74 @@ +class ImapToolsError(Exception): + """Base lib error""" + + +class MailboxFolderStatusValueError(ImapToolsError): + """Wrong folder status value error""" + + +class UnexpectedCommandStatusError(ImapToolsError): + """Unexpected status in IMAP command response""" + + def __init__(self, command_result: tuple, expected: str): + """ + :param command_result: imap command result + :param expected: expected command status + """ + self.command_result = command_result + self.expected = expected + + def __str__(self): + return 'Response status "{exp}" expected, but "{typ}" received. Data: {data}'.format( + exp=self.expected, typ=self.command_result[0], data=str(self.command_result[1])) + + +class MailboxFolderSelectError(UnexpectedCommandStatusError): + pass + + +class MailboxFolderCreateError(UnexpectedCommandStatusError): + pass + + +class MailboxFolderRenameError(UnexpectedCommandStatusError): + pass + + +class MailboxFolderDeleteError(UnexpectedCommandStatusError): + pass + + +class MailboxFolderStatusError(UnexpectedCommandStatusError): + pass + + +class MailboxLoginError(UnexpectedCommandStatusError): + pass + + +class MailboxLogoutError(UnexpectedCommandStatusError): + pass + + +class MailboxSearchError(UnexpectedCommandStatusError): + pass + + +class MailboxFetchError(UnexpectedCommandStatusError): + pass + + +class MailboxExpungeError(UnexpectedCommandStatusError): + pass + + +class MailboxDeleteError(UnexpectedCommandStatusError): + pass + + +class MailboxCopyError(UnexpectedCommandStatusError): + pass + + +class MailboxFlagError(UnexpectedCommandStatusError): + pass diff --git a/imap_tools/folder.py b/imap_tools/folder.py index 606d1e2..270c0f8 100644 --- a/imap_tools/folder.py +++ b/imap_tools/folder.py @@ -2,17 +2,32 @@ from . import imap_utf7 from .utils import check_command_status, quote, pairs_to_dict - - -class MailBoxFolderWrongStatusError(Exception): - """Wrong folder status error""" +from .errors import MailboxFolderStatusValueError, MailboxFolderSelectError, MailboxFolderCreateError, \ + MailboxFolderRenameError, MailboxFolderDeleteError, MailboxFolderStatusError + + +class MailBoxFolderStatusOptions: + """Valid mailbox folder status options""" + MESSAGES = 'MESSAGES' + RECENT = 'RECENT' + UIDNEXT = 'UIDNEXT' + UIDVALIDITY = 'UIDVALIDITY' + UNSEEN = 'UNSEEN' + all = ( + MESSAGES, RECENT, UIDNEXT, UIDVALIDITY, UNSEEN + ) + description = ( + (MESSAGES, "The number of messages in the mailbox"), + (RECENT, "The number of messages with the Recent flag set"), + (UIDNEXT, "The next unique identifier value of the mailbox"), + (UIDVALIDITY, "The unique identifier validity value of the mailbox"), + (UNSEEN, "The number of messages which do not have the Seen flag set"), + ) class MailBoxFolderManager: """Operations with mail box folders""" - folder_status_options = ('MESSAGES', 'RECENT', 'UIDNEXT', 'UIDVALIDITY', 'UNSEEN') - def __init__(self, mailbox): self.mailbox = mailbox self._current_folder = None @@ -28,7 +43,7 @@ def _encode_folder(folder: str or bytes) -> bytes: def set(self, folder: str or bytes): """Select current folder""" result = self.mailbox.box.select(self._encode_folder(folder)) - check_command_status('box.select', result) + check_command_status(result, MailboxFolderSelectError) self._current_folder = folder return result @@ -42,7 +57,7 @@ def create(self, folder: str or bytes): *Use email box delimiter to separate folders. Example for "|" delimiter: "folder|sub folder" """ result = self.mailbox.box._simple_command('CREATE', self._encode_folder(folder)) - check_command_status('CREATE', result) + check_command_status(result, MailboxFolderCreateError) return result def get(self): @@ -53,37 +68,33 @@ def rename(self, old_name: str or bytes, new_name: str or bytes): """Renemae folder from old_name to new_name""" result = self.mailbox.box._simple_command( 'RENAME', self._encode_folder(old_name), self._encode_folder(new_name)) - check_command_status('RENAME', result) + check_command_status(result, MailboxFolderRenameError) return result def delete(self, folder: str or bytes): """Delete folder""" result = self.mailbox.box._simple_command('DELETE', self._encode_folder(folder)) - check_command_status('DELETE', result) + check_command_status(result, MailboxFolderDeleteError) return result def status(self, folder: str or bytes, options: [str] or None = None) -> dict: """ Get the status of a folder :param folder: mailbox folder - :param options: [str] with values from MailBoxFolderManager.folder_status_options | None - for get all options - MESSAGES - The number of messages in the mailbox. - RECENT - The number of messages with the Recent flag set. - UIDNEXT - The next unique identifier value of the mailbox. - UIDVALIDITY - The unique identifier validity value of the mailbox. - UNSEEN - The number of messages which do not have the Seen flag set. + :param options: [str] with values from MailBoxFolderStatusOptions.all | None - for get all options :return: dict with available options keys """ command = 'STATUS' if not options: - options = self.folder_status_options - if not all((i in self.folder_status_options for i in options)): - raise MailBoxFolderWrongStatusError(str(options)) + options = tuple(MailBoxFolderStatusOptions.all) + for opt in options: + if opt not in MailBoxFolderStatusOptions.all: + raise MailboxFolderStatusValueError(str(opt)) status_result = self.mailbox.box._simple_command( command, self._encode_folder(folder), '({})'.format(' '.join(options))) - check_command_status(command, status_result) + check_command_status(status_result, MailboxFolderStatusError) result = self.mailbox.box._untagged_response(status_result[0], status_result[1], command) - check_command_status(command, result) + check_command_status(result, MailboxFolderStatusError) values = result[1][0].decode().split('(')[1].split(')')[0].split(' ') return {k: int(v) for k, v in pairs_to_dict(values).items() if str(v).isdigit()} diff --git a/imap_tools/mailbox.py b/imap_tools/mailbox.py index dcf70a6..329b7a5 100644 --- a/imap_tools/mailbox.py +++ b/imap_tools/mailbox.py @@ -1,8 +1,10 @@ import imaplib -from .message import MailMessage +from .message import MailMessage, MailMessageFlags from .folder import MailBoxFolderManager -from .utils import cleaned_uid_set, check_command_status, MessageFlags +from .utils import cleaned_uid_set, check_command_status +from .errors import MailboxLoginError, MailboxLogoutError, MailboxSearchError, MailboxFetchError, MailboxExpungeError, \ + MailboxDeleteError, MailboxCopyError, MailboxFlagError # Maximal line length when calling readline(). This is to prevent reading arbitrary length lines. imaplib._MAXLINE = 4 * 1024 * 1024 # 4Mb @@ -24,7 +26,7 @@ def _get_mailbox_client(self) -> imaplib.IMAP4: def login(self, username: str, password: str, initial_folder: str = 'INBOX'): result = self.box.login(username, password) - check_command_status('box.login', result) + check_command_status(result, MailboxLoginError) self.folder = self.folder_manager_class(self) self.folder.set(initial_folder) self.login_result = result @@ -32,7 +34,7 @@ def login(self, username: str, password: str, initial_folder: str = 'INBOX'): def logout(self): result = self.box.logout() - check_command_status('box.logout', result, expected='BYE') + check_command_status(result, MailboxLogoutError, expected='BYE') return result @staticmethod @@ -54,8 +56,10 @@ def fetch(self, criteria: str or bytes = 'ALL', charset: str = 'US-ASCII', limit :param headers_only: get only email headers (without text, html, attachments) :return generator: MailMessage """ + if headers_only: + raise NotImplementedError('headers_only does not work correctly and is disabled until fix, *you may help') search_result = self.box.search(charset, self._criteria_encoder(criteria, charset)) - check_command_status('box.search', search_result) + check_command_status(search_result, MailboxSearchError) # first element is string with email numbers through the gap message_id_set = search_result[1][0].decode().split(' ') if search_result[1][0] else () message_parts = "(BODY{}[{}] UID FLAGS)".format('' if mark_seen else '.PEEK', 'HEADER' if headers_only else '') @@ -64,7 +68,7 @@ def fetch(self, criteria: str or bytes = 'ALL', charset: str = 'US-ASCII', limit break # get message by id fetch_result = self.box.fetch(message_id, message_parts) - check_command_status('box.fetch', fetch_result) + check_command_status(fetch_result, MailboxFetchError) mail_message = self.email_message_class(fetch_result[1]) if miss_defect and mail_message.obj.defects: continue @@ -74,7 +78,7 @@ def fetch(self, criteria: str or bytes = 'ALL', charset: str = 'US-ASCII', limit def expunge(self) -> tuple: result = self.box.expunge() - check_command_status('box.expunge', result) + check_command_status(result, MailboxExpungeError) return result def delete(self, uid_list) -> (tuple, tuple) or None: @@ -87,7 +91,7 @@ def delete(self, uid_list) -> (tuple, tuple) or None: if not uid_str: return None store_result = self.box.uid('STORE', uid_str, '+FLAGS', r'(\Deleted)') - check_command_status('box.delete', store_result) + check_command_status(store_result, MailboxDeleteError) expunge_result = self.expunge() return store_result, expunge_result @@ -101,7 +105,7 @@ def copy(self, uid_list, destination_folder: str) -> tuple or None: if not uid_str: return None copy_result = self.box.uid('COPY', uid_str, destination_folder) - check_command_status('box.copy', copy_result) + check_command_status(copy_result, MailboxCopyError) return copy_result def move(self, uid_list, destination_folder: str) -> (tuple, tuple) or None: @@ -122,7 +126,7 @@ def flag(self, uid_list, flag_set: [str] or str, value: bool) -> (tuple, tuple) """ Set/unset email flags Do nothing on empty uid_list - Standard flags contains in MessageFlags.all + Standard flags contains in message.MailMessageFlags.all :return: None on empty uid_list, command results otherwise """ uid_str = cleaned_uid_set(uid_list) @@ -133,7 +137,7 @@ def flag(self, uid_list, flag_set: [str] or str, value: bool) -> (tuple, tuple) store_result = self.box.uid( 'STORE', uid_str, ('+' if value else '-') + 'FLAGS', '({})'.format(' '.join(('\\' + i for i in flag_set)))) - check_command_status('box.flag', store_result) + check_command_status(store_result, MailboxFlagError) expunge_result = self.expunge() return store_result, expunge_result @@ -142,7 +146,7 @@ def seen(self, uid_list, seen_val: bool) -> (tuple, tuple) or None: Mark email as read/unread This is shortcut for flag method """ - return self.flag(uid_list, MessageFlags.SEEN, seen_val) + return self.flag(uid_list, MailMessageFlags.SEEN, seen_val) def __enter__(self): return self diff --git a/imap_tools/message.py b/imap_tools/message.py index ec776d5..992cc37 100644 --- a/imap_tools/message.py +++ b/imap_tools/message.py @@ -8,6 +8,19 @@ from .utils import decode_value, parse_email_addresses, parse_email_date +class MailMessageFlags: + """Standard email message flags""" + SEEN = 'SEEN' + ANSWERED = 'ANSWERED' + FLAGGED = 'FLAGGED' + DELETED = 'DELETED' + DRAFT = 'DRAFT' + RECENT = 'RECENT' + all = ( + SEEN, ANSWERED, FLAGGED, DELETED, DRAFT, RECENT + ) + + class MailMessage: """The email message""" diff --git a/imap_tools/query.py b/imap_tools/query.py index a03e839..e5261b6 100644 --- a/imap_tools/query.py +++ b/imap_tools/query.py @@ -4,7 +4,7 @@ import functools import collections -from .utils import cleaned_uid_set, short_month_names, quote +from .utils import cleaned_uid_set, SHORT_MONTH_NAMES, quote class LogicOperator(collections.UserString): @@ -12,7 +12,7 @@ def __init__(self, *converted_strings, **unconverted_dicts): self.converted_strings = converted_strings for val in converted_strings: if not any(isinstance(val, t) for t in (str, collections.UserString)): - raise ValueError('Unexpected type "{}" for converted part, str like obj expected'.format(type(val))) + raise TypeError('Unexpected type "{}" for converted part, str like obj expected'.format(type(val))) self.converted_params = ParamConverter(unconverted_dicts).convert() if not any((self.converted_strings, self.converted_params)): raise ValueError('{} expects params'.format(self.__class__.__name__)) @@ -67,10 +67,10 @@ class Header: def __init__(self, name: str, value: str): if not isinstance(name, str): - raise ValueError('Header-name expected str value, "{}" received'.format(type(name))) + raise TypeError('Header-name expected str value, "{}" received'.format(type(name))) self.name = quote(name) if not isinstance(value, str): - raise ValueError('Header-value expected str value, "{}" received'.format(type(value))) + raise TypeError('Header-value expected str value, "{}" received'.format(type(value))) self.value = quote(value) def __str__(self): @@ -122,50 +122,50 @@ def convert(self) -> [str]: @classmethod def format_date(cls, value: datetime.date) -> str: """To avoid locale affects""" - return '{}-{}-{}'.format(value.day, short_month_names[value.month - 1], value.year) + return '{}-{}-{}'.format(value.day, SHORT_MONTH_NAMES[value.month - 1], value.year) @staticmethod def cleaned_str(key, value) -> str: if type(value) is not str: - raise ValueError('"{}" expected str value, "{}" received'.format(key, type(value))) + raise TypeError('"{}" expected str value, "{}" received'.format(key, type(value))) return str(value) @staticmethod def cleaned_date(key, value) -> datetime.date: if type(value) is not datetime.date: - raise ValueError('"{}" expected datetime.date value, "{}" received'.format(key, type(value))) + raise TypeError('"{}" expected datetime.date value, "{}" received'.format(key, type(value))) return value @staticmethod def cleaned_bool(key, value) -> bool: if type(value) is not bool: - raise ValueError('"{}" expected bool value, "{}" received'.format(key, type(value))) + raise TypeError('"{}" expected bool value, "{}" received'.format(key, type(value))) return bool(value) @staticmethod def cleaned_true(key, value) -> True: if value is not True: - raise ValueError('"{}" expected "True", "{}" received'.format(key, type(value))) + raise TypeError('"{}" expected "True", "{}" received'.format(key, type(value))) return True @staticmethod def cleaned_uint(key, value) -> int: if type(value) is not int or int(value) < 0: - raise ValueError('"{}" expected int value >= 0, "{}" received'.format(key, type(value))) + raise TypeError('"{}" expected int value >= 0, "{}" received'.format(key, type(value))) return int(value) @staticmethod def cleaned_uid(key, value) -> str: try: uid_set = cleaned_uid_set(value) - except ValueError as e: - raise ValueError('{} parse error: {}'.format(key, str(e))) + except TypeError as e: + raise TypeError('{} parse error: {}'.format(key, str(e))) return uid_set @staticmethod def cleaned_header(key, value) -> H: if not isinstance(value, H): - raise ValueError('"{}" expected Header (H) value, "{}" received'.format(key, type(value))) + raise TypeError('"{}" expected Header (H) value, "{}" received'.format(key, type(value))) return value def convert_answered(self, key, value): diff --git a/imap_tools/utils.py b/imap_tools/utils.py index 850faab..2cccde1 100644 --- a/imap_tools/utils.py +++ b/imap_tools/utils.py @@ -4,47 +4,50 @@ from email.utils import getaddresses from email.header import decode_header -short_month_names = ('Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun', 'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec') +SHORT_MONTH_NAMES = ('Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun', 'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec') def cleaned_uid_set(uid_set: str or [str] or iter) -> str: """ - Prepare set of uid for use in commands: delete/copy/move/seen + Prepare set of uid for use in IMAP commands uid_set may be: str, that is comma separated uids Iterable, that contains str uids - Generator with "fetch" name, implicitly gets all non-empty uids + Generator with "fetch" name, implicitly gets all uids """ + # str if type(uid_set) is str: + if re.search(r'^(\d+,)*\d+$', uid_set): # *optimization for already good str + return uid_set uid_set = uid_set.split(',') + # Generator if inspect.isgenerator(uid_set) and getattr(uid_set, '__name__', None) == 'fetch': uid_set = tuple(msg.uid for msg in uid_set if msg.uid) + # Iterable try: uid_set_iter = iter(uid_set) except TypeError: - raise ValueError('Wrong uid type: "{}"'.format(type(uid_set))) + raise TypeError('Wrong uid_set arg type: "{}"'.format(type(uid_set))) + # check uid types for uid in uid_set_iter: if type(uid) is not str: - raise ValueError('uid "{}" is not string'.format(str(uid))) + raise TypeError('uid "{}" is not string'.format(str(uid))) if not uid.strip().isdigit(): - raise ValueError('Wrong uid: "{}"'.format(uid)) + raise TypeError('Wrong uid: "{}"'.format(uid)) return ','.join((i.strip() for i in uid_set)) -class UnexpectedCommandStatusError(Exception): - """Unexpected status in response""" - - -def check_command_status(command, command_result, expected='OK'): +def check_command_status(command_result: tuple, exception: type, expected='OK'): """ - Check that command responses status equals status - If not, raises UnexpectedCommandStatusError + Check that IMAP command responses status equals status + If not, raise specified + :param command_result: imap command result + :param exception: exception subclass of UnexpectedCommandStatusError, that raises + :param expected: expected command status """ typ, data = command_result[0], command_result[1] if typ != expected: - raise UnexpectedCommandStatusError( - 'Response status for command "{command}" == "{typ}", "{exp}" expected, data: {data}'.format( - command=command, typ=typ, data=str(data), exp=expected)) + raise exception(command_result=command_result, expected=expected) def decode_value(value: bytes or str, encoding=None) -> str: @@ -81,7 +84,7 @@ def parse_email_addresses(raw_header: str) -> (dict,): def parse_email_date(value: str) -> datetime.datetime: """Parsing the date described in rfc2822""" - match = re.search(r'(?P\d{1,2}\s+(' + '|'.join(short_month_names) + r')\s+\d{4})\s+' + + match = re.search(r'(?P\d{1,2}\s+(' + '|'.join(SHORT_MONTH_NAMES) + r')\s+\d{4})\s+' + r'(?P