You should have received a copy of the GNU General Public License
along with notmuch. If not, see <http://www.gnu.org/licenses/>.
-Copyright 2010 Sebastian Spaeth <Sebastian@SSpaeth.de>'
+Copyright 2010-2011 Sebastian Spaeth <Sebastian@SSpaeth.de>
"""
from notmuch.database import Database, Query
from notmuch.message import Messages, Message
from notmuch.thread import Threads, Thread
from notmuch.tag import Tags
-from notmuch.globals import nmlib, STATUS, NotmuchError
+from notmuch.globals import (
+ nmlib,
+ STATUS,
+ NotmuchError,
+ OutOfMemoryError,
+ ReadOnlyDatabaseError,
+ XapianError,
+ FileError,
+ FileNotEmailError,
+ DuplicateMessageIdError,
+ NullPointerError,
+ TagTooLongError,
+ UnbalancedFreezeThawError,
+ UnbalancedAtomicError,
+ NotInitializedError
+)
from notmuch.version import __VERSION__
__LICENSE__ = "GPL v3+"
__AUTHOR__ = 'Sebastian Spaeth <Sebastian@SSpaeth.de>'
def _assert_db_is_initialized(self):
"""Raises a NotmuchError in case self._db is still None"""
if self._db is None:
- raise NotmuchError(STATUS.NOT_INITIALIZED)
+ raise NotmuchError.get_subclass_exc(STATUS.NOT_INITIALIZED)
def create(self, path):
"""Creates a new notmuch database
res = Database._open(_str(path), mode)
if res is None:
- raise NotmuchError(
- message="Could not open the specified database")
+ raise NotmuchError(message="Could not open the specified database")
self._db = res
def get_path(self):
self._assert_db_is_initialized()
status = nmlib.notmuch_database_begin_atomic(self._db)
if status != STATUS.SUCCESS:
- raise NotmuchError(status)
+ raise NotmuchError.get_subclass_exc(status)
return status
def end_atomic(self):
self._assert_db_is_initialized()
status = nmlib.notmuch_database_end_atomic(self._db)
if status != STATUS.SUCCESS:
- raise NotmuchError(status)
+ raise NotmuchError.get_subclass_exc(status)
return status
def get_directory(self, path):
# we got an absolute path
if not path.startswith(self.get_path()):
# but its initial components are not equal to the db path
- raise NotmuchError(STATUS.FILE_ERROR,
+ raise NotmuchError.get_subclass_exc(STATUS.FILE_ERROR,
message="Database().get_directory() called "
"with a wrong absolute path.")
abs_dirpath = path
byref(msg_p))
if not status in [STATUS.SUCCESS, STATUS.DUPLICATE_MESSAGE_ID]:
- raise NotmuchError(status)
+ raise NotmuchError.get_subclass_exc(status)
#construct Message() and return
msg = Message(msg_p, self)
self._assert_db_is_initialized()
tags_p = Database._get_all_tags(self._db)
if tags_p == None:
- raise NotmuchError(STATUS.NULL_POINTER)
+ raise NotmuchError.get_subclass_exc(STATUS.NULL_POINTER)
return Tags(tags_p, self)
def create_query(self, querystring):
(too little memory)
"""
if db.db_p is None:
- raise NotmuchError(STATUS.NOT_INITIALIZED)
+ raise NotmuchError.get_subclass_exc(STATUS.NOT_INITIALIZED)
# create reference to parent db to keep it alive
self._db = db
# create query, return None if too little mem available
query_p = Query._create(db.db_p, _str(querystr))
if query_p is None:
- raise NotmuchError(STATUS.NULL_POINTER)
+ raise NotmuchError.get_subclass_exc(STATUS.NULL_POINTER)
self._query = query_p
def set_sort(self, sort):
been initialized.
"""
if self._query is None:
- raise NotmuchError(STATUS.NOT_INITIALIZED)
+ raise NotmuchError.get_subclass_exc(STATUS.NOT_INITIALIZED)
self.sort = sort
nmlib.notmuch_query_set_sort(self._query, sort)
* :attr:`STATUS`.NULL_POINTER if search_threads failed
"""
if self._query is None:
- raise NotmuchError(STATUS.NOT_INITIALIZED)
+ raise NotmuchError.get_subclass_exc(STATUS.NOT_INITIALIZED)
threads_p = Query._search_threads(self._query)
* :attr:`STATUS`.NULL_POINTER if search_messages failed
"""
if self._query is None:
- raise NotmuchError(STATUS.NOT_INITIALIZED)
+ raise NotmuchError.get_subclass_exc(STATUS.NOT_INITIALIZED)
msgs_p = Query._search_messages(self._query)
if msgs_p is None:
- raise NotmuchError(STATUS.NULL_POINTER)
+ raise NotmuchError.get_subclass_exc(STATUS.NULL_POINTER)
return Messages(msgs_p, self)
* :attr:`STATUS`.NOT_INITIALIZED if query is not inited
"""
if self._query is None:
- raise NotmuchError(STATUS.NOT_INITIALIZED)
+ raise NotmuchError.get_subclass_exc(STATUS.NOT_INITIALIZED)
return Query._count_messages(self._query)
def _assert_dir_is_initialized(self):
"""Raises a NotmuchError(:attr:`STATUS`.NOT_INITIALIZED) if dir_p is None"""
if self._dir_p is None:
- raise NotmuchError(STATUS.NOT_INITIALIZED)
+ raise NotmuchError.get_subclass_exc(STATUS.NOT_INITIALIZED)
def __init__(self, path, dir_p, parent):
"""
if status == STATUS.SUCCESS:
return
#fail with Exception otherwise
- raise NotmuchError(status)
+ raise NotmuchError.get_subclass_exc(status)
def get_mtime(self):
"""Gets the mtime value of this directory in the database
def next(self):
if self._files_p is None:
- raise NotmuchError(STATUS.NOT_INITIALIZED)
+ raise NotmuchError.get_subclass_exc(STATUS.NOT_INITIALIZED)
if not nmlib.notmuch_filenames_valid(self._files_p):
self._files_p = None
for file in files: print file
"""
if self._files_p is None:
- raise NotmuchError(STATUS.NOT_INITIALIZED)
+ raise NotmuchError.get_subclass_exc(STATUS.NOT_INITIALIZED)
i = 0
while nmlib.notmuch_filenames_valid(self._files_p):
argument to receive a human readable string"""
STATUS.__name__ = 'STATUS'
-
class NotmuchError(Exception):
- def __init__(self, status=None, message=None):
- """Is initiated with a (notmuch.STATUS[,message=None])"""
+ """Is initiated with a (notmuch.STATUS[, message=None]). It will not
+ return an instance of the class NotmuchError, but a derived instance
+ of a more specific Error Message, e.g. OutOfMemoryError. Each status
+ but SUCCESS has a corresponding subclassed Exception."""
+
+ @classmethod
+ def get_subclass_exc(cls, status, message=None):
+ """Returns a fine grained Exception() type,detailing the error status"""
+ subclasses = {
+ STATUS.OUT_OF_MEMORY: OutOfMemoryError,
+ STATUS.READ_ONLY_DATABASE: ReadOnlyDatabaseError,
+ STATUS.XAPIAN_EXCEPTION: XapianError,
+ STATUS.FILE_ERROR: FileError,
+ STATUS.FILE_NOT_EMAIL: FileNotEmailError,
+ STATUS.DUPLICATE_MESSAGE_ID: DuplicateMessageIdError,
+ STATUS.NULL_POINTER: NullPointerError,
+ STATUS.TAG_TOO_LONG: TagTooLongError,
+ STATUS.UNBALANCED_FREEZE_THAW: UnbalancedFreezeThawError,
+ STATUS.UNBALANCED_ATOMIC: UnbalancedAtomicError,
+ STATUS.NOT_INITIALIZED: NotInitializedError
+ }
+ assert 0 < status <= len(subclasses)
+ return subclasses[status](status, message)
+
+ def __init__(self, status, message=None):
self.status = status
self.message = message
else:
return 'Unknown error'
+# List of Subclassed exceptions that correspond to STATUS values and are
+# subclasses of NotmuchError:
+class OutOfMemoryError(NotmuchError):
+ pass
+class ReadOnlyDatabaseError(NotmuchError):
+ pass
+class XapianError(NotmuchError):
+ pass
+class FileError(NotmuchError):
+ pass
+class FileNotEmailError(NotmuchError):
+ pass
+class DuplicateMessageIdError(NotmuchError):
+ pass
+class NullPointerError(NotmuchError):
+ pass
+class TagTooLongError(NotmuchError):
+ pass
+class UnbalancedFreezeThawError(NotmuchError):
+ pass
+class UnbalancedAtomicError(NotmuchError):
+ pass
+class NotInitializedError(NotmuchError):
+ pass
+
+
def _str(value):
"""Ensure a nicely utf-8 encoded string to pass to libnotmuch
projects / notmuch-old / commitdiff