6
\$\begingroup\$

Back again for more punishment improvement advice.

Basically, stemming from my dislike of how standard imaplib.IMAP4 mailbox interaction commands functioned, I made an extension library for imaplib, called imaplibext (on PyPI now), which takes the existing IMAP4 commands that have UID-equivalents, and actually use the UID-equivalent commands. While this is a very evil approach to a very simple problem, it makes code a little easier to understand for number of different projects I work on.

I'm looking for a general review here, as well as Py2/Py3 cross-compatibility concerns. While I don't have any specific concerns as everything seems to behave well and work as I expect it to, I'm sure I'm doing stupid things that I'm overlooking.

Suggestions for improvement of this already-functioning code is welcome. The actual code part for this is attached below, but the full package itself can be found on the github repository as well.


import imaplib
import socket
from typing import Union, Tuple, AnyStr, List, Any
import sys


class IMAP4(imaplib.IMAP4):
    def __init__(self, host='', port=imaplib.IMAP4_PORT, timeout=None, maxbytes=None):
        # type: (AnyStr, int) -> None
        # Override standard __init__ - we need to add a timeout option and a maxbytes option.

        # This timeout option is used below in the 'open' function override.
        if timeout:
            socket.setdefaulttimeout(timeout)

        # This maxbytes option is used below to override the max bytes allowed to be returned from UID commands and
        # others if defined.  Otherwise it leaves it at the 10000 default.
        if maxbytes:
            imaplib._MAXLINE = maxbytes

        imaplib.IMAP4.__init__(self, host, port)
        return  # PEP compliance

    def copy(self, message_set, new_mailbox):
        # type: (AnyStr, AnyStr) -> Tuple[AnyStr, list]
        """Copy 'message_set' messages onto end of 'new_mailbox'.

        (typ, [data]) = <instance>.copy(message_set, new_mailbox)
        """

        # conn.uid('COPY', message_set, new_mailbox)
        return self.uid('COPY', message_set, new_mailbox)

    def fetch(self, message_set, message_parts):
        # type: (AnyStr, AnyStr) -> Tuple[AnyStr, List[Tuple[Any]]]
        """Fetch (parts of) messages, using UID values.

        (typ, [data, ...]) = <instance>.fetch(message_set, message_parts)

        'message_parts' should be a string of selected parts
        enclosed in parentheses, eg: "(UID BODY[TEXT])".

        'data' are tuples of message part envelope and data.
        """

        # conn.uid('FETCH', msgset, parts)
        return self.uid('FETCH', message_set, message_parts)

    def search(self, charset, *criteria):
        # type: (Union[AnyStr, None], Union[AnyStr, tuple]) -> Tuple[AnyStr, list]
        """Search mailbox for matching messages.

        (typ, [data]) = <instance>.search(charset, criterion, ...)

        'data' is space separated list of matching message UID numbers.
        If UTF8 is enabled, charset MUST be None.
        """

        # conn.uid('SEARCH', charset, criteria)
        return self.uid('SEARCH', charset, " ".join(criteria))

    def sort(self, sort_criteria, charset, *search_criteria):
        # type: (AnyStr, Union[AnyStr, None], Union[AnyStr,tuple]) -> Tuple[AnyStr, list]
        """IMAP4rev1 extension SORT command.

        (typ, [data]) = <instance>.sort(sort_criteria, charset, search_criteria, ...)
        """

        # conn.uid('SORT', '(SORT CRITERION)', 'CHARSET', 'SEARCH_CRITERIA')

        # Preprocess the search_criterion tuple - make sure all string data is split up in order
        # to make each component of the argument as its own tuple item, instead of strings with
        # spaces.
        _search_criterion = []
        for criterion in search_criteria:
            criterion = str(criterion)
            if ' ' in criterion:
                for subcriterion in criterion.split():
                    _search_criterion.append(subcriterion)
            else:
                _search_criterion.append(criterion)

        search_criteria = tuple(list(_search_criterion))

        # Preprocess the 'sort criteria' provided - make sure it's all in parentheses.
        while True:
            if sort_criteria[0] != '(':
                sort_criteria = '(' + sort_criteria
                continue

            if sort_criteria[len(sort_criteria) - 1] != ')':
                sort_criteria += ')'
                continue

            break

        # Charset is a required argument, so if we give a charset of "None", we can assume UTF-8 here.
        if not charset:
            charset = 'UTF-8'

        return self.uid('SORT', sort_criteria, charset, ' '.join(search_criteria))

    def store(self, message_set, command, flags):
        # type: (AnyStr, AnyStr, AnyStr) -> Tuple[AnyStr, list]
        """Alters flag dispositions for messages in mailbox, using UID values.

        (typ, [data]) = <instance>.store(message_set, command, flags)
        """

        # conn.uid('STORE', msg_uid, '-FLAGS', '(\Seen)')
        return self.uid('STORE', message_set, command, flags)

    def thread(self, threading_algorithm, charset, *search_criteria):
        # type: (AnyStr, Union[AnyStr, None], Union[AnyStr, Tuple]) -> Tuple[AnyStr, Any, list]
        """IMAPrev1 extension THREAD command.

        (type, [data]) = <instance>.thread(threading_algorithm, charset, search_criteria, ...)
        """

        # Preprocess the search_criterion tuple - make sure all string data is split up in order
        # to make each component of the argument as its own tuple item, instead of strings with
        # spaces.
        _search_criterion = []
        for criterion in search_criteria:
            criterion = str(criterion)
            if ' ' in criterion:
                for subcriterion in criterion.split():
                    _search_criterion.append(subcriterion)
            else:
                _search_criterion.append(criterion)

        search_criteria = tuple(list(_search_criterion))

        if not charset:
            charset = 'UTF-8'

        # conn.uid('THREAD', threading_algorithm, charset, search criterion)
        return self.uid('THREAD', threading_algorithm, charset, search_criteria)


# noinspection PyPep8Naming
class IMAP4_SSL(imaplib.IMAP4_SSL):
    def __init__(self, host='', port=imaplib.IMAP4_PORT, timeout=None, maxbytes=None,
                 keyfile=None, certfile=None, ssl_context=None):
        # type: (AnyStr, int, int, any, any, any) -> None
        # Override standard __init__ - we need to add a timeout option.
        # This timeout option is used below in the 'open' function override.
        self.timeout = timeout
        if timeout:
            socket.setdefaulttimeout(timeout)

        # This maxbytes option is used below to override the max bytes allowed to be returned from UID commands and
        # others if defined.  Otherwise it leaves it at the 10000 default.
        if maxbytes:
            imaplib._MAXLINE = maxbytes

        if sys.version_info.major < 3:
            if ssl_context:
                print("Warning: Defining `ssl_context` is not supported in "
                      "Python 2's IMAP_SSL implementation.")
            imaplib.IMAP4_SSL.__init__(self, host, port, keyfile, certfile)
        else:
            imaplib.IMAP4_SSL.__init__(self, host, port, keyfile, certfile, ssl_context)

        return  # PEP compliance

    def copy(self, message_set, new_mailbox):
        # type: (AnyStr, AnyStr) -> Tuple[AnyStr, list]
        """Copy 'message_set' messages onto end of 'new_mailbox'.

        (typ, [data]) = <instance>.copy(message_set, new_mailbox)
        """

        # conn.uid('COPY', message_set, new_mailbox)
        return self.uid('COPY', message_set, new_mailbox)

    def fetch(self, message_set, message_parts):
        # type: (AnyStr, AnyStr) -> Tuple[AnyStr, List[Tuple[Any]]]
        """Fetch (parts of) messages, using UID values.

        (typ, [data, ...]) = <instance>.fetch(message_set, message_parts)

        'message_parts' should be a string of selected parts
        enclosed in parentheses, eg: "(UID BODY[TEXT])".

        'data' are tuples of message part envelope and data.
        """

        # conn.uid('FETCH', msgset, parts)
        return self.uid('FETCH', message_set, message_parts)

    def search(self, charset, *criteria):
        # type: (Union[AnyStr, None], Union[AnyStr, tuple]) -> Tuple[AnyStr, list]
        """Search mailbox for matching messages.

        (typ, [data]) = <instance>.search(charset, criterion, ...)

        'data' is space separated list of matching message UID numbers.
        If UTF8 is enabled, charset MUST be None.
        """

        # conn.uid('SEARCH', charset, criteria)
        return self.uid('SEARCH', charset, " ".join(criteria))

    def sort(self, sort_criteria, charset, *search_criteria):
        # type: (AnyStr, Union[AnyStr, None], Union[AnyStr,tuple]) -> Tuple[AnyStr, list]
        """IMAP4rev1 extension SORT command.

        (typ, [data]) = <instance>.sort(sort_criteria, charset, search_criteria, ...)
        """

        # conn.uid('SORT', '(SORT CRITERION)', 'CHARSET', 'SEARCH_CRITERIA')

        # Preprocess the search_criterion tuple - make sure all string data is split up in order
        # to make each component of the argument as its own tuple item, instead of strings with
        # spaces.
        _search_criterion = []
        for criterion in search_criteria:
            criterion = str(criterion)
            if ' ' in criterion:
                for subcriterion in criterion.split():
                    _search_criterion.append(subcriterion)
            else:
                _search_criterion.append(criterion)

        search_criteria = tuple(list(_search_criterion))

        # Preprocess the 'sort criteria' provided - make sure it's all in parentheses.
        while True:
            if sort_criteria[0] != '(':
                sort_criteria = '(' + sort_criteria
                continue

            if sort_criteria[len(sort_criteria) - 1] != ')':
                sort_criteria += ')'
                continue

            break

        # Charset is a required argument, so if we give a charset of "None", we can assume UTF-8 here.
        if not charset:
            charset = 'UTF-8'

        return self.uid('SORT', sort_criteria, charset, ' '.join(search_criteria))

    def store(self, message_set, command, flags):
        # type: (AnyStr, AnyStr, AnyStr) -> Tuple[AnyStr, list]
        """Alters flag dispositions for messages in mailbox, using UID values.

        (typ, [data]) = <instance>.store(message_set, command, flags)
        """

        # conn.uid('STORE', msg_uid, '-FLAGS', '(\Seen)')
        return self.uid('STORE', message_set, command, flags)

    def thread(self, threading_algorithm, charset, *search_criteria):
        # type: (AnyStr, Union[AnyStr, None], Union[AnyStr, Tuple]) -> Tuple[AnyStr, Any, list]
        """IMAPrev1 extension THREAD command.

        (type, [data]) = <instance>.thread(threading_algorithm, charset, search_criteria, ...)
        """

        # Preprocess the search_criterion tuple - make sure all string data is split up in order
        # to make each component of the argument as its own tuple item, instead of strings with
        # spaces.
        _search_criterion = []
        for criterion in search_criteria:
            criterion = str(criterion)
            if ' ' in criterion:
                for subcriterion in criterion.split():
                    _search_criterion.append(subcriterion)
            else:
                _search_criterion.append(criterion)

        search_criteria = tuple(list(_search_criterion))

        if not charset:
            charset = 'UTF-8'

        # conn.uid('THREAD', threading_algorithm, charset, search criterion)
        return self.uid('THREAD', threading_algorithm, charset, search_criteria)
\$\endgroup\$

2 Answers 2

3
\$\begingroup\$

Your rather complicated part

    while True:
        if sort_criteria[0] != '(':
            sort_criteria = '(' + sort_criteria
            continue

        if sort_criteria[len(sort_criteria) - 1] != ')':
            sort_criteria += ')'
            continue

        break

may be written as

    if sort_criteria[0] != '(':
        sort_criteria = '(' + sort_criteria

    if sort_criteria[len(sort_criteria) - 1] != ')':
        sort_criteria += ')'

or - in the more Pythonic way -

    if not sort_criteria.startswith('('):
        sort_criteria = '(' + sort_criteria

    if not sort_criteria.endswith(')'):
        sort_criteria += ')'
\$\endgroup\$
4
  • 2
    \$\begingroup\$ I would also point out that negative indexes also work: sort_criteria[-1]. \$\endgroup\$ Aug 26, 2017 at 15:53
  • 1
    \$\begingroup\$ Are you sure that your "more pythonic way" will work? Because IIRC, *sort_criteria returns a tuple, not a string. I'm inclined to believe that your last recommendation will explode as that's string manipulation and not tuple manipulation. That being said, I could adapt your first suggested rewrite as a negative-index approach. \$\endgroup\$ Aug 26, 2017 at 15:55
  • \$\begingroup\$ Oh, so I didn't carefully read your code. I'm glad that at least one my suggestion is useful for you. \$\endgroup\$
    – MarianD
    Aug 26, 2017 at 16:45
  • \$\begingroup\$ @MarianD Actually I'm blind heh. Rereading my code, the manipulation can work as a string, but eh, we'll see if this works without exploding. \$\endgroup\$ Sep 1, 2017 at 3:44
3
\$\begingroup\$
  • The type-hints are incomplete for IMAP4.__init__ and IMAP4_SSL.__init__.
  • Try to use typing.List and typing.Tuple instead of list and tuple. The former allow you to subtype a little further, for eg. List[str].
  • Use super() instead of explicit calls: imaplib.IMAP4.__init__(self, host, port), that would help in case of multiple inheritance.
  • Union[AnyStr, None] can be replaced with Optional[AnyStr].
  • There's no need of this check: if ' ' in criterion. Simply split and call extend on _search_criterion instead of a loop and list.append: _search_criterion.extend(criterion.split()).
  • The list() call in tuple(list(_search_criterion)) is redundant.
  • The from based imports should come after simple imports, plus make sure the import groups itself are sorted. Check Google's style guide and PEP 8.

    import imaplib
    import socket
    import sys
    from typing import Union, Tuple, AnyStr, List, Any  
    
  • I am not sure about return in __init__ being PEP compliant.

\$\endgroup\$
5
  • \$\begingroup\$ Same to me for your last bullet point. Maybe @ThomasWard can provide a bit more explanation about that comment? \$\endgroup\$ Aug 26, 2017 at 18:20
  • \$\begingroup\$ @MathiasEttinger extremely literal interpretation of PEP rules, removed. \$\endgroup\$ Sep 1, 2017 at 3:46
  • \$\begingroup\$ @AshwiniChaudhary I tried to do a super call with super(); however that never worked. Googling the issue, it was found that IMAP4 needs the direct init call, not a super(); due to whatever issues seem to be in the system. At least, for Python 2 to work properly. Once Py2 dies permanently, I can start making it fully modernized. \$\endgroup\$ Sep 1, 2017 at 3:47
  • \$\begingroup\$ Namely, regarding the call to the explicit init calls, I did it this way because imaplib is an old-style class per this SO post and doesn't work proper with super(). This is something I found as well when trying to develop this extension library. \$\endgroup\$ Sep 1, 2017 at 3:52
  • \$\begingroup\$ Going in order from first to last, I'll make some notes. Type hints are now completed. I've switched to using typing.* now, though at the time of my initial module writing, there was a PyCharm type hinting bug that I was working around. Super call won't work, imaplib is old-style, see my other comment. Union involving None is now moved to Optional. I've updated _search_criterion handling. Removed the list() call in the search criterion tuple conversion. Reordered the imports. Also removed the return in __init__. \$\endgroup\$ Sep 1, 2017 at 3:57

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.