cyimap.py

# -*- encoding=utf-8 -*-

from copy import copy

import imaplib

class CyrusIMAPMixin(object):
  """
    Mixin for imaplib.IMAP4 and imaplib.IMAP4SSL 
    
    to implement cyrus-specific commands, including administrative commands.
    
  """
  ANN_CYRUS_PARTITION = '/vendor/cmu/cyrus-imapd/partition'

	def __new__(cls, *args, **kwargs):
		cls.Commands = copy(cls.Commands)
		Commands.update({
			'RECONSTRUCT'  : ('AUTH',),
			'DUMP'         : ('AUTH',), # To check admin status
			'ID'           : ('AUTH',), # Only one ID allowed in non auth mode
			'GETANNOTATION': ('AUTH',),
			'SETANNOTATION': ('AUTH',)
		})

		return super(cls, CyrusIMAPMixin).__new__(*args, **kwargs)

	def create(self, mailbox, partition=None):
    """
      Create a new `mailbox` with optional `partition`.
    """
		if partition:
			return self._simple_command('CREATE', mailbox, '(PARTITION {})'.format(self._checkquote(partition)))
		else:
			return self._simple_command('CREATE', mailbox)

	def getannotation(self, mailbox, pattern, domain=None):
		"""
      Get all annotations from `mailbox` matching `pattern`.
      
      Use annotation domain `domain`. If `domain` is not specified, it defaults to `value.shared`.
    """
    domain = domain or 'value.shared'
		typ, dat = self._simple_command('GETANNOTATION', mailbox, self._quote(pattern), self._quote(domain))
		return self._untagged_response(typ, dat, 'ANNOTATION')

	def getannotation_parsed(self, mailbox, pattern, domain=None):
    """
      Same as `.getannotation()`, but returned parsed response arguments.
    """
		typ, dat = self.getannotation(mailbox, pattern, domain=domain)
		if not dat or not any(dat):
			return typ, []
		else:
			values = [self._parse_response(value) for value in dat]
			return typ, values

	def setannotation(self, mailbox, desc, value, domain=None):
		value = value or 'NIL'
		domain = domain or 'value.shared'
		typ, dat = self._simple_command('SETANNOTATION', mailbox, desc, '({}, {})'.format(self._quote(domain), self._quote(value)))
		return self._untagged_response(typ, dat, 'ANNOTATION')

	def get_partition(self, mailbox):
    """
      Get the partition name for `mailbox`.
    """
		typ, dat = self.getannotation_parsed(mailbox, self.ANN_CYRUS_PARTITION)
		domain, value = self._parse_response(dat[0][2])
		return value

	def _parse_response(self, response, accept_brackets=True):
    """
      Parse annotation responses into individual tokens.
      
      if `accept_brackets` is set, we also accept bracket-quoted arguments.
      Bracked-quoted multi-arguments (e.g. annotation values) are not
      parsed recursively.      
    """
		last_slash = False
		tokens = []
		offset = 0
		endquote = ''
		while len(response) > offset:
			token = ''
			if response[offset] == '"':
				endquote = '"'
			elif accept_brackets and response[offset] == '(':
				endquote = ')'
			else:
				endquote = ' '
				token = response[offset]

			offset += 1
			while len(response) > offset and not last_slash and response[offset] != endquote:
				if endquote == '"' and response[offset] == '\\':
					last_slash = True
				else:
					token += response[offset]
				offset += 1

			if len(response) < offset or (endquote != ' ' and response[offset] != endquote):
				raise ValueError('Unterminated quoted string "{}"'.format(token))
			if endquote != ' ':
				offset += 1
			if len(response) > offset:
				if response[offset] != ' ':
					raise ValueError('Expected separating space, got "{}"'.format(response[offset:]))
				offset += 1

			tokens.append(token)
		return tokens

  #
  # Extension to make IMAP4 objects available as 
  # context managers. Issues `.logout()` on exit.
  # Any errors from logout are discarded.
	def __enter__(self):
		return self

	def __exit__(self, exc, val, tb):
		try:
			self.logout()
		except self.error:
			# XXX - log me
			pass


class CyrusIMAP4(CyrusIMAPMixin, imaplib.IMAP4):
	pass


class CyrusIMAP4SSL(CyrusIMAPMixin, imaplib.IMAP4_SSL):
	pass