Skip to content

Instantly share code, notes, and snippets.

@thomaswieland
Created February 17, 2018 13:56
Show Gist options
  • Save thomaswieland/3cac92843896040b11c4635f7bf61cfb to your computer and use it in GitHub Desktop.
Save thomaswieland/3cac92843896040b11c4635f7bf61cfb to your computer and use it in GitHub Desktop.
Python: IMAP IDLE with imaplib2
import imaplib2, time
from threading import *
# This is the threading object that does all the waiting on
# the event
class Idler(object):
def __init__(self, conn):
self.thread = Thread(target=self.idle)
self.M = conn
self.event = Event()
def start(self):
self.thread.start()
def stop(self):
# This is a neat trick to make thread end. Took me a
# while to figure that one out!
self.event.set()
def join(self):
self.thread.join()
def idle(self):
# Starting an unending loop here
while True:
# This is part of the trick to make the loop stop
# when the stop() command is given
if self.event.isSet():
return
self.needsync = False
# A callback method that gets called when a new
# email arrives. Very basic, but that's good.
def callback(args):
if not self.event.isSet():
self.needsync = True
self.event.set()
# Do the actual idle call. This returns immediately,
# since it's asynchronous.
self.M.idle(callback=callback)
# This waits until the event is set. The event is
# set by the callback, when the server 'answers'
# the idle call and the callback function gets
# called.
self.event.wait()
# Because the function sets the needsync variable,
# this helps escape the loop without doing
# anything if the stop() is called. Kinda neat
# solution.
if self.needsync:
self.event.clear()
self.dosync()
# The method that gets called when a new email arrives.
# Replace it with something better.
def dosync(self):
print "Got an event!"
# Had to do this stuff in a try-finally, since some testing
# went a little wrong.....
try:
# Set the following two lines to your creds and server
M = imaplib2.IMAP4_SSL("mail.example.com")
M.login("mylogin","mypassword")
# We need to get out of the AUTH state, so we just select
# the INBOX.
M.select("INBOX")
# Start the Idler thread
idler = Idler(M)
idler.start()
# Because this is just an example, exit after 1 minute.
time.sleep(1*60)
finally:
# Clean up.
idler.stop()
idler.join()
M.close()
# This is important!
M.logout()
@andydoc
Copy link

andydoc commented Dec 7, 2020

@jaynery I have written my own, less polished code for my own use case and at the beginning of the equivalent of dosync() I do idle_done(), at the end I do idle()

@ydf
Copy link

ydf commented Aug 17, 2021


    def get_unseen_messages(self) -> list:
        typ, data = self.M.search(None, 'UNSEEN')
        if typ == u'OK':
            return data[0].split()
        raise ValueError("Get unseen messages Error")

    def get_message(self, message_number):
        typ, data = self.M.fetch(message_number, 'RFC822')
        if typ == 'OK':
            return email.message_from_bytes(data[0][1])
        raise Exception

    def dosync(self):
        message_numbers = self.get_unseen_messages()
        message_numbers.reverse()
        for message_number in message_numbers:
            mail_obj = self.get_message(message_number)
            print(mail_obj)

@bfontaine
Copy link

bfontaine commented Jan 5, 2023

If you can use a library, this one has built-in IDLE support:
https://imapclient.readthedocs.io/en/2.3.1/advanced.html

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment