I am attempting to have a python script that constantly monitors a gmail account for new emails. Using IMAPClient, it opens two imap connections, one in idle mode. Whenever a new message is received, the connection in idle mode tells the other connection that it should fetch new mail. The non-idle connection will fetch the mail, perform some processing, then archive the email.
The problem comes when I have many emails arriving in a short period of time, more than a few in a minute. In this case, I get an AssertionError. Below is a minimal example to reproduce the error. In addition to the imap connection, it also opens an smtp connection so that it can send the emails to itself. It will usually fail with the AssertionError at some point after 5-7 emails have been sent. The AssertionError comes in the call to idle_check.
A few short comments on running the code. It does not use OAuth, and so gmail's must be set to allow less secure apps. The "username" and "password" fields at the bottom of the script must be set. The script will also archive any emails that are currently in the inbox, and so it should not be run on a primary email account.
#!/usr/bin/env python3
import smtplib
import imapclient
import email
import threading
import time
class Server(object):
def __init__(self,username,password):
self.username = username
self.password = password
def start(self):
self.stop_running = threading.Event()
self.has_mail = threading.Event()
threading.Thread(target=self._idle).start()
threading.Thread(target=self._poll).start()
print('Listening for messages now')
def _idle(self):
imap_idle = self.imap_connect()
while not self.stop_running.is_set():
imap_idle.idle()
for i in range(600):
try:
if imap_idle.idle_check(1):
self.has_mail.set()
except AssertionError as e:
self.stop_running.set()
raise
imap_idle.idle_done()
imap_idle.noop()
def _poll(self):
imap_poll = self.imap_connect()
self.process_unread(imap_poll)
while True:
if self.has_mail.wait(1):
self.has_mail.clear()
self.process_unread(imap_poll)
if self.stop_running.is_set():
return
def imap_connect(self):
imap = imapclient.IMAPClient('imap.gmail.com',use_uid=True,ssl=True)
imap.login(self.username,self.password)
imap.select_folder('INBOX')
return imap
def process_unread(self, imap):
imap.select_folder('INBOX')
messages = imap.search()
if messages:
imap.copy(messages,'[Gmail]/All Mail')
imap.delete_messages(messages)
def smtp_connect(self):
smtp = smtplib.SMTP('smtp.gmail.com',587)
smtp.ehlo()
smtp.starttls()
smtp.ehlo()
smtp.login(self.username,self.password)
return smtp
def send(self,recipient,subject='',body=''):
headers = ['from: ' + self.username,
'subject: ' + subject,
'to: ' + recipient,
'mime-version: 1.0',
'content-type: text/html']
headers = '\r\n'.join(headers)
self.smtp_connect().sendmail(self.username,recipient,headers+'\r\n\r\n'+body)
def main():
username = 'username@gmail.com'
password = 'password'
s = Server(username, password)
s.start()
for i in range(8):
if s.stop_running.is_set():
break
print('Sending message',i)
s.send(username,
'Test Message'.format(i),
'Body {}'.format(i))
time.sleep(15)
if __name__=='__main__':
main()
The error messsage given is as follows. The text variable at the time of error is sometimes b'XISTS' and sometimes b' FLAGS (\\Seen))'.
Exception in thread Thread-1:
Traceback (most recent call last):
File "/usr/lib/python3.4/threading.py", line 920, in _bootstrap_inner
self.run()
File "/usr/lib/python3.4/threading.py", line 868, in run
self._target(*self._args, **self._kwargs)
File "./emailer.py", line 31, in _idle
if imap_idle.idle_check(1):
File "/path/to/the/venv/lib/python3.4/site-packages/imapclient/imapclient.py", line 519, in idle_check
resps.append(_parse_untagged_response(line))
File "/path/to/the/venv/lib/python3.4/site-packages/imapclient/imapclient.py", line 1093, in _parse_untagged_response
assert text.startswith(b'* ')
AssertionError
I am running this with Python 3.4.0, using IMAPClient 0.13. This is being run in a clean virtualenv, with no other libraries installed. Any assistance would be quite appreciated.
