zev/imbox
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added missing documentation for all supported query keyword arguments. Fixes #124. Added Pycharm directory to .gitignore.

Browse Source 
Fixed var names in documentation of query keywords

moved Messages and Imbox to their own modules, imported Imbox.imbox into __init__.py and put it in __all__. fixes #130.

clarified in documentation and Imbox.messages logging that, unless a folder is specified in the kwargs to Imbox.messages, the returned messages will be from the inbox.  In the documentation this is accomplished exclusively by the var names. fixes #128.

amended `8df7d7c` to reflect manual changes made to `README.rst` in current master, but also added `inbox_` to several var names to make that explicit in the documentation.  Added flags to messages returned by `fetch_email_by_uid`, using the new function `parse_flags` in `parser.py`.  Fixes #126.

added TODO back into query.py
This commit is contained in:
zevav
2018-07-25 08:22:04 -04:00
parent 06aa4e054b
commit 9cffb51a81
7 changed files with 173 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@@ -33,4 +33,4 @@ example.*
example.py
# PyCharm
.idea/
.idea/

33
README.rst
View File

@@ -36,40 +36,45 @@ Usage
ssl=True,
ssl_context=None,
starttls=False) as imbox:
# Get all folders
status, folders_with_additional_info = imbox.folders()
# Gets all messages from the inbox
all_messages = imbox.messages()
all_inbox_messages = imbox.messages()
# Unread messages
unread_messages = imbox.messages(unread=True)
unread_inbox_messages = imbox.messages(unread=True)
# Flagged messages
inbox_flagged_messages = imbox.messages(flagged=True)
# Un-flagged messages
inbox_unflagged_messages = imbox.messages(unflagged=True)
# Messages sent FROM
messages_from = imbox.messages(sent_from='sender@example.org')
inbox_messages_from = imbox.messages(sent_from='sender@example.org')
# Messages sent TO
messages_to = imbox.messages(sent_to='receiver@example.org')
inbox_messages_to = imbox.messages(sent_to='receiver@example.org')
# Messages received before specific date
messages_received_before = imbox.messages(date__lt=datetime.date(2018, 7, 31))
inbox_messages_received_before = imbox.messages(date__lt=datetime.date(2018, 7, 31))
# Messages received after specific date
messages_received_after = imbox.messages(date__gt=datetime.date(2018, 7, 30))
inbox_messages_received_after = imbox.messages(date__gt=datetime.date(2018, 7, 30))
# Messages received on a specific date
messages_received_on_date = imbox.messages(date__on=datetime.date(2018, 7, 30))
inbox_messages_received_on_date = imbox.messages(date__on=datetime.date(2018, 7, 30))
# Messages from a specific folder
messages_from_folder = imbox.messages(folder='Social')
# Messages whose subjects contain a string
messages_subject_christmas = imbox.messages(subject='Christmas')
# Messages whose subjects contain a string
inbox_messages_subject_christmas = imbox.messages(subject='Christmas')
for uid, message in all_messages:
for uid, message in all_inbox_messages:
# Every message is an object with the following keys
message.sent_from
@@ -129,7 +134,7 @@ Usage
# mark the message as read
imbox.mark_seen(uid)
Changelog

127
imbox/__init__.py
View File

@@ -1,131 +1,8 @@
from imbox.imap import ImapTransport
from imbox.parser import parse_email, fetch_email_by_uid
from imbox.query import build_search_query
import logging
logger = logging.getLogger(__name__)
from imbox.imbox import Imbox
__version_info__ = (0, 9, 5)
__version__ = '.'.join([str(x) for x in __version_info__])
class Imbox:
__all__ = ['Imbox']
def __init__(self, hostname, username=None, password=None, ssl=True,
port=None, ssl_context=None, policy=None, starttls=False):
self.server = ImapTransport(hostname, ssl=ssl, port=port,
ssl_context=ssl_context, starttls=starttls)
self.hostname = hostname
self.username = username
self.password = password
self.parser_policy = policy
self.connection = self.server.connect(username, password)
logger.info("Connected to IMAP Server with user {username} on {hostname}{ssl}".format(
hostname=hostname, username=username, ssl=(" over SSL" if ssl or starttls else "")))
def __enter__(self):
return self
def __exit__(self, type, value, traceback):
self.logout()
def logout(self):
self.connection.close()
self.connection.logout()
logger.info("Disconnected from IMAP Server {username}@{hostname}".format(
hostname=self.hostname, username=self.username))
def mark_seen(self, uid):
logger.info("Mark UID {} with \\Seen FLAG".format(int(uid)))
self.connection.uid('STORE', uid, '+FLAGS', '(\\Seen)')
def mark_flag(self, uid):
logger.info("Mark UID {} with \\Flagged FLAG".format(int(uid)))
self.connection.uid('STORE', uid, '+FLAGS', '(\\Flagged)')
def delete(self, uid):
logger.info("Mark UID {} with \\Deleted FLAG and expunge.".format(int(uid)))
self.connection.expunge()
def copy(self, uid, destination_folder):
logger.info("Copy UID {} to {} folder".format(int(uid), str(destination_folder)))
return self.connection.uid('COPY', uid, destination_folder)
def move(self, uid, destination_folder):
logger.info("Move UID {} to {} folder".format(int(uid), str(destination_folder)))
if self.copy(uid, destination_folder):
self.delete(uid)
def messages(self, **kwargs):
folder = kwargs.get('folder', False)
msg = ""
if folder:
self.connection.select(folder)
msg = " from folder '{}'".format(folder)
logger.info("Fetch list of messages{}".format(msg))
return Messages(connection=self.connection,
parser_policy=self.parser_policy,
**kwargs)
def folders(self):
return self.connection.list()
class Messages:
def __init__(self,
connection,
parser_policy,
**kwargs):
self.connection = connection
self.parser_policy = parser_policy
self.kwargs = kwargs
self._uid_list = self._query_uids(**kwargs)
logger.debug("Fetch all messages for UID in {}".format(self._uid_list))
def _fetch_email(self, uid):
return fetch_email_by_uid(uid=uid,
connection=self.connection,
parser_policy=self.parser_policy)
def _query_uids(self, **kwargs):
query_ = build_search_query(**kwargs)
message, data = self.connection.uid('search', None, query_)
if data[0] is None:
return []
return data[0].split()
def _fetch_email_list(self):
for uid in self._uid_list:
yield uid, self._fetch_email(uid)
def __repr__(self):
if len(self.kwargs) > 0:
return 'Messages({})'.format('\n'.join('{}={}'.format(key, value)
for key, value in self.kwargs.items()))
return 'Messages(ALL)'
def __iter__(self):
return self._fetch_email_list()
def __next__(self):
return self
def __len__(self):
return len(self._uid_list)
def __getitem__(self, index):
uids = self._uid_list[index]
if not isinstance(uids, list):
uid = uids
return uid, self._fetch_email(uid)
return [(uid, self._fetch_email(uid))
for uid in uids]

72
imbox/imbox.py Normal file
View File

@@ -0,0 +1,72 @@
from imbox.imap import ImapTransport
from imbox.messages import Messages
import logging
logger = logging.getLogger(__name__)
class Imbox:
def __init__(self, hostname, username=None, password=None, ssl=True,
port=None, ssl_context=None, policy=None, starttls=False):
self.server = ImapTransport(hostname, ssl=ssl, port=port,
ssl_context=ssl_context, starttls=starttls)
self.hostname = hostname
self.username = username
self.password = password
self.parser_policy = policy
self.connection = self.server.connect(username, password)
logger.info("Connected to IMAP Server with user {username} on {hostname}{ssl}".format(
hostname=hostname, username=username, ssl=(" over SSL" if ssl or starttls else "")))
def __enter__(self):
return self
def __exit__(self, type, value, traceback):
self.logout()
def logout(self):
self.connection.close()
self.connection.logout()
logger.info("Disconnected from IMAP Server {username}@{hostname}".format(
hostname=self.hostname, username=self.username))
def mark_seen(self, uid):
logger.info("Mark UID {} with \\Seen FLAG".format(int(uid)))
self.connection.uid('STORE', uid, '+FLAGS', '(\\Seen)')
def mark_flag(self, uid):
logger.info("Mark UID {} with \\Flagged FLAG".format(int(uid)))
self.connection.uid('STORE', uid, '+FLAGS', '(\\Flagged)')
def delete(self, uid):
logger.info("Mark UID {} with \\Deleted FLAG and expunge.".format(int(uid)))
self.connection.expunge()
def copy(self, uid, destination_folder):
logger.info("Copy UID {} to {} folder".format(int(uid), str(destination_folder)))
return self.connection.uid('COPY', uid, destination_folder)
def move(self, uid, destination_folder):
logger.info("Move UID {} to {} folder".format(int(uid), str(destination_folder)))
if self.copy(uid, destination_folder):
self.delete(uid)
def messages(self, **kwargs):
folder = kwargs.get('folder', False)
if folder:
self.connection.select(folder)
msg = " from folder '{}'".format(folder)
else:
msg = " from inbox"
logger.info("Fetch list of messages{}".format(msg))
return Messages(connection=self.connection,
parser_policy=self.parser_policy,
**kwargs)
def folders(self):
return self.connection.list()

62
imbox/messages.py Normal file
View File

@@ -0,0 +1,62 @@
from imbox.parser import fetch_email_by_uid
from imbox.query import build_search_query
import logging
logger = logging.getLogger(__name__)
class Messages:
def __init__(self,
connection,
parser_policy,
**kwargs):
self.connection = connection
self.parser_policy = parser_policy
self.kwargs = kwargs
self._uid_list = self._query_uids(**kwargs)
logger.debug("Fetch all messages for UID in {}".format(self._uid_list))
def _fetch_email(self, uid):
return fetch_email_by_uid(uid=uid,
connection=self.connection,
parser_policy=self.parser_policy)
def _query_uids(self, **kwargs):
query_ = build_search_query(**kwargs)
message, data = self.connection.uid('search', None, query_)
if data[0] is None:
return []
return data[0].split()
def _fetch_email_list(self):
for uid in self._uid_list:
yield uid, self._fetch_email(uid)
def __repr__(self):
if len(self.kwargs) > 0:
return 'Messages({})'.format('\n'.join('{}={}'.format(key, value)
for key, value in self.kwargs.items()))
return 'Messages(ALL)'
def __iter__(self):
return self._fetch_email_list()
def __next__(self):
return self
def __len__(self):
return len(self._uid_list)
def __getitem__(self, index):
uids = self._uid_list[index]
if not isinstance(uids, list):
uid = uids
return uid, self._fetch_email(uid)
return [(uid, self._fetch_email(uid))
for uid in uids]

18
imbox/parser.py
View File

@@ -1,8 +1,10 @@
import imaplib
import io
import re
import email
import base64
import quopri
import sys
import time
from datetime import datetime
from email.header import decode_header
@@ -129,15 +131,27 @@ def decode_content(message):
def fetch_email_by_uid(uid, connection, parser_policy):
message, data = connection.uid('fetch', uid, '(BODY.PEEK[])')
message, data = connection.uid('fetch', uid, '(BODY.PEEK[] FLAGS)')
logger.debug("Fetched message for UID {}".format(int(uid)))
raw_email = data[0][1]
raw_headers, raw_email = data[0]
email_object = parse_email(raw_email, policy=parser_policy)
flags = parse_flags(raw_headers.decode())
email_object.__dict__['flags'] = flags
return email_object
def parse_flags(headers):
"""Copied from https://github.com/girishramnani/gmail/blob/master/gmail/message.py"""
if len(headers) == 0:
return []
if sys.version_info[0] == 3:
headers = bytes(headers, "ascii")
return list(imaplib.ParseFlags(headers))
def parse_email(raw_email, policy=None):
if isinstance(raw_email, bytes):
raw_email = str_encode(raw_email, 'utf-8', errors='ignore')

2
imbox/query.py
View File

@@ -1,6 +1,6 @@
import datetime
import logging
# TODO - Validate query arguments
-# TODO - Validate query arguments
logger = logging.getLogger(__name__)