--- a/VirtualMailManager/catchall.py Mon Mar 24 19:22:04 2014 +0200
+++ /dev/null Thu Jan 01 00:00:00 1970 +0000
@@ -1,181 +0,0 @@
-# -*- coding: UTF-8 -*-
-# Copyright (c) 2012 - 2014, martin f. krafft
-# See COPYING for distribution information.
-"""
- VirtualMailManager.catchall
- ~~~~~~~~~~~~~~~~~~~~~~~~~~~
-
- Virtual Mail Manager's CatchallAlias class to manage domain catch-all
- aliases.
-
- This is heavily based on (more or less a copy of) the Alias class, because
- fundamentally, catchall aliases are aliases, but without a localpart.
- While Alias could potentially derive from CatchallAlias to reuse some of
- the functionality, it's probably not worth it. I found no sensible way to
- derive CatchallAlias from Alias, or at least none that would harness the
- powers of polymorphism.
-
- Yet, we reuse the AliasError exception class, which makes sense.
-"""
-
-from VirtualMailManager.domain import get_gid
-from VirtualMailManager.emailaddress import \
- EmailAddress, DestinationEmailAddress as DestAddr
-from VirtualMailManager.errors import AliasError as AErr
-from VirtualMailManager.ext.postconf import Postconf
-from VirtualMailManager.pycompat import all
-from VirtualMailManager.constants import \
- ALIAS_EXCEEDS_EXPANSION_LIMIT, NO_SUCH_ALIAS, NO_SUCH_DOMAIN
-
-
-_ = lambda msg: msg
-cfg_dget = lambda option: None
-
-
-class CatchallAlias(object):
- """Class to manage domain catch-all aliases."""
- __slots__ = ('_domain', '_dests', '_gid', '_dbh')
-
- def __init__(self, dbh, domain):
- self._domain = domain
- self._dbh = dbh
- self._gid = get_gid(self._dbh, self.domain)
- if not self._gid:
- raise AErr(_(u"The domain '%s' does not exist.") %
- self.domain, NO_SUCH_DOMAIN)
- self._dests = []
-
- self._load_dests()
-
- def _load_dests(self):
- """Loads all known destination addresses into the _dests list."""
- dbc = self._dbh.cursor()
- dbc.execute('SELECT destination FROM catchall WHERE gid = %s',
- (self._gid,))
- dests = dbc.fetchall()
- if dbc.rowcount > 0:
- self._dests.extend(DestAddr(dest[0], self._dbh) for dest in dests)
- dbc.close()
-
- def _check_expansion(self, count_new):
- """Checks the current expansion limit of the alias."""
- postconf = Postconf(cfg_dget('bin.postconf'))
- limit = long(postconf.read('virtual_alias_expansion_limit'))
- dcount = len(self._dests)
- failed = False
- if dcount == limit or dcount + count_new > limit:
- failed = True
- errmsg = _(
-u"""Cannot add %(count_new)i new destination(s) to catch-all alias for
-domain '%(domain)s'. Currently this alias expands into %(count)i/%(limit)i
-recipients. %(count_new)i additional destination(s) will render this alias
-unusable.
-Hint: Increase Postfix' virtual_alias_expansion_limit""")
- elif dcount > limit:
- failed = True
- errmsg = _(
-u"""Cannot add %(count_new)i new destination(s) to catch-all alias for domain
-'%(domain)s'. This alias already exceeds its expansion limit \
-(%(count)i/%(limit)i).
-So its unusable, all messages addressed to this alias will be bounced.
-Hint: Delete some destination addresses.""")
- if failed:
- raise AErr(errmsg % {'domain': self._domain, 'count': dcount,
- 'limit': limit, 'count_new': count_new},
- ALIAS_EXCEEDS_EXPANSION_LIMIT)
-
- def _delete(self, destinations=None):
- """Delete one ore multiple destinations from the catchall alias, if
- ``destinations`` is not ``None``. If ``destinations`` is None, the
- catchall alias with all its destination addresses will be deleted.
-
- """
- dbc = self._dbh.cursor()
- if not destinations:
- dbc.execute('DELETE FROM catchall WHERE gid = %s', (self._gid,))
- else:
- dbc.executemany('DELETE FROM catchall WHERE gid = %d AND '
- 'destination = %%s' % self._gid,
- ((str(dest),) for dest in destinations))
- if dbc.rowcount > 0:
- self._dbh.commit()
- dbc.close()
-
- def __len__(self):
- """Returns the number of destinations of the catchall alias."""
- return len(self._dests)
-
- @property
- def domain(self):
- """The Alias' domain."""
- return self._domain
-
- def add_destinations(self, destinations, warnings=None):
- """Adds the `EmailAddress`es from *destinations* list to the
- destinations of the catchall alias.
-
- Destinations, that are already assigned to the alias, will be
- removed from *destinations*. When done, this method will return
- a set with all destinations, that were saved in the database.
- """
- destinations = set(destinations)
- assert destinations and \
- all(isinstance(dest, EmailAddress) for dest in destinations)
- if not warnings is None:
- assert isinstance(warnings, list)
- duplicates = destinations.intersection(set(self._dests))
- if duplicates:
- destinations.difference_update(set(self._dests))
- if not warnings is None:
- warnings.extend(duplicates)
- if not destinations:
- return destinations
- self._check_expansion(len(destinations))
- dbc = self._dbh.cursor()
- dbc.executemany("INSERT INTO catchall (gid, destination) "
- "VALUES (%d, %%s)" % self._gid,
- ((str(destination),) for destination in destinations))
- self._dbh.commit()
- dbc.close()
- self._dests.extend(destinations)
- return destinations
-
- def del_destinations(self, destinations, warnings=None):
- """Deletes the specified ``destinations`` from the catchall alias."""
- destinations = set(destinations)
- assert destinations and \
- all(isinstance(dest, EmailAddress) for dest in destinations)
- if not warnings is None:
- assert isinstance(warnings, list)
- if not self._dests:
- raise AErr(_(u"There are no catch-all aliases defined for "
- u"domain '%s'.") % self._domain, NO_SUCH_ALIAS)
- unknown = destinations.difference(set(self._dests))
- if unknown:
- destinations.intersection_update(set(self._dests))
- if not warnings is None:
- warnings.extend(unknown)
- if not destinations:
- raise AErr(_(u"No suitable destinations left to remove from the "
- u"catch-all alias of domain '%s'.") % self._domain,
- NO_SUCH_ALIAS)
- self._delete(destinations)
- for destination in destinations:
- self._dests.remove(destination)
-
- def get_destinations(self):
- """Returns an iterator for all destinations of the catchall alias."""
- if not self._dests:
- raise AErr(_(u"There are no catch-all aliases defined for "
- u"domain '%s'.") % self._domain, NO_SUCH_ALIAS)
- return iter(self._dests)
-
- def delete(self):
- """Deletes all catchall destinations for the domain."""
- if not self._dests:
- raise AErr(_(u"There are no catch-all aliases defined for "
- u"domain '%s'.") % self._domain, NO_SUCH_ALIAS)
- self._delete()
- del self._dests[:]
-
-del _, cfg_dget