imaplib_idle.py

import imaplib

# Source idea based on https://bugs.python.org/file48790/0001-Add-IDLE-support-to-imaplib.patch
# modified cases and provided demo

# 01.2021 by Markus Ullmann <[email protected]>
# License: WTFPL

# Version: 1.0 Initial version

class Monkey:
    
    @property
    def already_patched(self):
        return 'IDLE' in imaplib.Commands

    def done(self):
        """Exit IDLE state"""
        if self.state == 'IDLE':
            self._command('DONE')
            self.state = 'SELECTED'
            return None
        else:
            self.error('DONE outside of IDLE')
            return None

    def idle(self):
        """Enter or continue an IDLE state"""

        if 'IDLE' not in self.capabilities:
            raise imaplib.IMAP4.error("Server does not support IDLE")

        self._command('IDLE')
        if self._get_response() is None and self.continuation_response == b'idling':
            self.state = 'IDLE'
            while self.state == 'IDLE':
                response = self._get_response()
                if b'OK' in response and b'IDLE terminated' in response:
                    self.state = 'SELECTED'
                    continue
                if b'OK' in response:
                    # just keeping connection alive, ignore
                    print("keep alive, ignored:", response)
                    continue
                elif b'EXISTS' in response: 
                    # Existing message count changed
                    yield ('EXISTS', self.untagged_responses['EXISTS'])
                elif b'EXPUNGE' in response:
                    # Some message got deleted
                    yield ('EXPUNGE', self.untagged_responses['EXPUNGE'])
                elif b'RECENT' in response:
                    # When messages get fetched, also return count of unread ones
                    yield ('RECENT', self.untagged_responses['RECENT'])
        else:
            raise self.error("Failed to IDLE")


    def patch(self):
        if self.already_patched:
            return

        imaplib.Commands['DONE'] = ('IDLE', )
        imaplib.Commands['IDLE'] = ('SELECTED', )
        imaplib.IMAP4.done = Monkey.done
        imaplib.IMAP4.idle = Monkey.idle


def demo():
    Monkey().patch()

    server = "example.imap.server"
    username = "testuser"
    password = "testpass"

    print("connect")
    conn = imaplib.IMAP4(server)
    print("tls")
    conn.starttls()
    print("login")
    conn.login(username, password)

    print("open INBOX")
    response = conn.select("INBOX")
    status, count = response
    msgcount = int(count[0].decode("utf-8"))
    # handle old ones elsewhere, just wait for new ones with IDLE in example

    print("idle")
    while True:
        idle_channel = conn.idle()
        while True:    
            event, state in next(idle_channel)
            if event == "EXISTS":
                # fetch msgs and handle, yet ignore all expunges or recents changes
                break
        conn.done()
        # conn.search() .....

        # if satisfied or 30 minutes are reached (per RFC 2177 throw connection away every 30 mins)
        # break

if __name__ == '__main__':
    demo()