How to use imaplib2 - 10 common examples

To help you get started, we’ve selected a few imaplib2 examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github bamthomas / aioimaplib / tests / test_imapserver_imaplib2.py View on Github external
def test_idle(self):
        imap_client = yield from self.login_user('user', 'pass', select=True, lib=imaplib2.IMAP4)
        idle_callback = Mock()
        self.loop.run_in_executor(None, functools.partial(imap_client.idle, callback=idle_callback))
        yield from asyncio.wait_for(get_imapconnection('user').wait(imapserver.IDLE), 1)

        self.loop.run_in_executor(None, functools.partial(imap_receive, Mail(to=['user'], mail_from='me', subject='hello')))

        yield from asyncio.wait_for(get_imapconnection('user').wait(imapserver.SELECTED), 1)
        yield from asyncio.sleep(0.1) # eurk hate sleeps but I don't know how to wait for the lib to receive end of IDLE
        idle_callback.assert_called_once()
github bamthomas / aioimaplib / tests / test_imapserver_imaplib2.py View on Github external
def test_login_twice(self):
        with self.assertRaises(imaplib2.IMAP4.error) as expected:
            imap_client = yield from self.login_user('user', 'pass', lib=imaplib2.IMAP4)

            yield from asyncio.wait_for(
                self.loop.run_in_executor(None, functools.partial(imap_client.login, 'user', 'pass')), 1)

        self.assertEqual(expected.exception.args, ('command LOGIN illegal in state AUTH',))
github bamthomas / aioimaplib / tests / test_imapserver_imaplib2.py View on Github external
def test_login_twice(self):
        with self.assertRaises(imaplib2.IMAP4.error) as expected:
            imap_client = yield from self.login_user('user', 'pass', lib=imaplib2.IMAP4)

            yield from asyncio.wait_for(
                self.loop.run_in_executor(None, functools.partial(imap_client.login, 'user', 'pass')), 1)

        self.assertEqual(expected.exception.args, ('command LOGIN illegal in state AUTH',))
github goerz / procimap / ProcImap / ProcImap.py View on Github external
self.mailbox.close()
            sys.exit()
        else:
            while True:
                try:
                    unseen_msgs = self.mailbox.get_unseen_uids()
                    if len(unseen_msgs) > 0:
                        self._process_uids(unseen_msgs)
                        self.mailbox.expunge()
                    try:
                        self.mailbox.server.idle(TIMEOUT)
                    except:
                        log("IDLE caused exception. Sleeping for %s seconds" \
                                                                      % TIMEOUT)
                        time.sleep(TIMEOUT)
                except imaplib.IMAP4.abort, data:
                    log("In run(): caught exception: %s" % str(data))
                    self.mailbox.reconnect()
                    log("Reconnected mailbox")
                    # TODO: check if this could end in an infinite try-except
github assembl / assembl / assembl / processes / imaplib2_source_reader.py View on Github external
try:
                if self.source.message_ok_to_import(message_string):
                    (email_object, dummy, error) = self.source.parse_email(message_string)
                    if error:
                        raise ReaderError(error)
                    self.source.db.add(email_object)
                else:
                    print "Skipped message with imap id %s (bounce or vacation message)" % (email_id)
                # print "Setting self.source.last_imported_email_uid to "+email_id
                self.source.last_imported_email_uid = email_id
                self.source.db.commit()
            finally:
                self.source = ContentSource.get(self.source.id)
        except IMAP4.abort as e:
            raise IrrecoverableError(e)
        except IMAP4.error as e:
            raise ClientError(e)
github assembl / assembl / assembl / models / mail.py View on Github external
    @staticmethod
    def do_import_content(mbox, only_new=True):
        mbox = mbox.db.merge(mbox)
        session = mbox.db
        session.add(mbox)
        if mbox.use_ssl:
            mailbox = IMAP4_SSL(host=mbox.host.encode('utf-8'), port=mbox.port)
        else:
            mailbox = IMAP4(host=mbox.host.encode('utf-8'), port=mbox.port)
        if 'STARTTLS' in mailbox.capabilities:
            # Always use starttls if server supports it
            mailbox.starttls()
        mailbox.login(mbox.username, mbox.password)
        mailbox.select(mbox.folder)

        command = "ALL"
        search_status = None

        email_ids = None
        if only_new and mbox.last_imported_email_uid:
            command = "(UID %s:*)" % mbox.last_imported_email_uid

            search_status, search_result = mailbox.uid('search', None, command)
            # print "UID searched with: "+ command + ", got result "+repr(search_status)+" and found "+repr(search_result)
            email_ids = search_result[0].split()
github assembl / assembl / assembl / processes / imaplib2_source_reader.py View on Github external
if not message_string:
                raise ClientError()
            try:
                if self.source.message_ok_to_import(message_string):
                    (email_object, dummy, error) = self.source.parse_email(message_string)
                    if error:
                        raise ReaderError(error)
                    self.source.db.add(email_object)
                else:
                    print "Skipped message with imap id %s (bounce or vacation message)" % (email_id)
                # print "Setting self.source.last_imported_email_uid to "+email_id
                self.source.last_imported_email_uid = email_id
                self.source.db.commit()
            finally:
                self.source = ContentSource.get(self.source.id)
        except IMAP4.abort as e:
            raise IrrecoverableError(e)
        except IMAP4.error as e:
            raise ClientError(e)
github assembl / assembl / assembl / processes / imaplib2_source_reader.py View on Github external
if not is_ok((search_status,)):
                    raise ReaderError(search_result)
                #print "UID searched with: "+ command + ", got result "+repr(search_status)+" and found "+repr(search_result)
                email_ids = search_result[0].split()

            if len(email_ids):
                print "Processing messages from IMAP: %d "% (len(email_ids))
                for email_id in email_ids:
                    self.import_email(email_id)
                    if self.status != ReaderStatus.READING:
                        break
            else:
                print "No IMAP messages to process"
            self.successful_read()
            self.set_status(ReaderStatus.PAUSED)
        except IMAP4.abort as e:
            raise IrrecoverableError(e)
        except IMAP4.error as e:
            raise ClientError(e)
github assembl / assembl / assembl / processes / imaplib2_source_reader.py View on Github external
#print "UID searched with: "+ command + ", got result "+repr(search_status)+" and found "+repr(search_result)
                email_ids = search_result[0].split()

            if len(email_ids):
                print "Processing messages from IMAP: %d "% (len(email_ids))
                for email_id in email_ids:
                    self.import_email(email_id)
                    if self.status != ReaderStatus.READING:
                        break
            else:
                print "No IMAP messages to process"
            self.successful_read()
            self.set_status(ReaderStatus.PAUSED)
        except IMAP4.abort as e:
            raise IrrecoverableError(e)
        except IMAP4.error as e:
            raise ClientError(e)
github assembl / assembl / assembl / lib / email.py View on Github external
def connection(self):
        """Open a connection to the IMAP server, login and select a mailbox."""
        if self._connection is None:
            imap_cls = imaplib2.IMAP4_SSL if self.ssl else imaplib2.IMAP4
            self._connection = imap_cls(self.host, self.port)
            self._connection.login(self.username, self.password)
            code, [msg] = self._connection.select(self.mailbox)
            if not code == 'OK':
                raise MailboxSelectionError(msg)
            self._mailbox_selected = True
        return self._connection

imaplib2

A threaded Python IMAP4 client.

MIT
Latest version published 3 years ago

Package Health Score

63 / 100
Full package analysis

Similar packages