VMM/Transport: reworked Transport class.
Use assertions for argument checks. Removed methods getID() and
getTransport(). This values are now accessible through the read-only
attributes id and transport.
VMM/{Account,Domain}: adjusted to modifications in Transport class.
A few code cleanups.
--- a/VirtualMailManager/Account.py Mon Mar 01 04:46:46 2010 +0000
+++ b/VirtualMailManager/Account.py Mon Mar 01 05:31:43 2010 +0000
@@ -11,9 +11,14 @@
from VirtualMailManager.MailLocation import MailLocation
from VirtualMailManager.Transport import Transport
+
+_ = lambda msg: msg
+
+
class Account(object):
"""Class to manage e-mail accounts."""
- __slots__ = ('_addr','_base','_gid','_mid','_passwd','_tid','_uid','_dbh')
+ __slots__ = ('_addr', '_base', '_gid', '_mid', '_passwd', '_tid', '_uid',
+ '_dbh')
def __init__(self, dbh, address, password=None):
self._dbh = dbh
@@ -45,7 +50,7 @@
dbc = self._dbh.cursor()
dbc.execute(
"SELECT uid, mid, tid FROM users WHERE gid=%s AND local_part=%s",
- self._gid, self._addr._localpart)
+ self._gid, self._addr.localpart)
result = dbc.fetchone()
dbc.close()
if result is not None:
@@ -55,11 +60,11 @@
return False
def _setAddr(self):
- dom = Domain(self._dbh, self._addr._domainname)
+ dom = Domain(self._dbh, self._addr.domainname)
self._gid = dom.getID()
if self._gid == 0:
raise AccE(_(u"The domain “%s” doesn't exist.") %
- self._addr._domainname, ERR.NO_SUCH_DOMAIN)
+ self._addr.domainname, ERR.NO_SUCH_DOMAIN)
self._base = dom.getDir()
self._tid = dom.getTransportID()
@@ -105,7 +110,7 @@
def __aliaseCount(self):
dbc = self._dbh.cursor()
q = "SELECT COUNT(destination) FROM alias WHERE destination = '%s'"\
- %self._addr
+ % self._addr
dbc.execute(q)
a_count = dbc.fetchone()[0]
dbc.close()
@@ -141,7 +146,7 @@
self._prepare(maillocation)
sql = "INSERT INTO users (local_part, passwd, uid, gid, mid, tid,\
smtp, pop3, imap, %s) VALUES ('%s', '%s', %d, %d, %d, %d, %s, %s, %s, %s)" % (
- sieve_col, self._addr._localpart, self._passwd, self._uid,
+ sieve_col, self._addr.localpart, self._passwd, self._uid,
self._gid, self._mid, self._tid, smtp, pop3, imap, sieve)
dbc = self._dbh.cursor()
dbc.execute(sql)
@@ -162,7 +167,7 @@
dbc.execute('UPDATE users SET passwd = %s WHERE uid = %s',
value, self._uid)
elif what == 'transport':
- self._tid = Transport(self._dbh, transport=value).getID()
+ self._tid = Transport(self._dbh, transport=value).id
dbc.execute('UPDATE users SET tid = %s WHERE uid = %s',
self._tid, self._uid)
else:
@@ -202,7 +207,7 @@
MailLocation(self._dbh,
mid=info['maildir']).getMailLocation())
info['transport'] = Transport(self._dbh,
- tid=info['transport']).getTransport()
+ tid=info['transport']).transport
return info
def getAliases(self):
@@ -241,9 +246,11 @@
dbc.close()
raise AccE(
_(u"There are %(count)d aliases with the destination address\
- “%(address)s”.") %{'count': a_count, 'address': self._addr}, ERR.ALIAS_PRESENT)
+ “%(address)s”.") % {'count': a_count, 'address': self._addr},
+ ERR.ALIAS_PRESENT)
dbc.close()
+
def getAccountByID(uid, dbh):
try:
uid = long(uid)
@@ -264,3 +271,5 @@
info = dict(zip(keys, info))
return info
+
+del _
--- a/VirtualMailManager/Domain.py Mon Mar 01 04:46:46 2010 +0000
+++ b/VirtualMailManager/Domain.py Mon Mar 01 05:31:43 2010 +0000
@@ -15,11 +15,14 @@
MAILDIR_CHARS = '0123456789abcdefghijklmnopqrstuvwxyz'
+_ = lambda x: x
class Domain(object):
"""Class to manage e-mail domains."""
- __slots__ = ('_basedir','_domaindir','_id','_name','_transport','_dbh')
+ __slots__ = ('_basedir', '_domaindir', '_id', '_name', '_transport',
+ '_dbh')
+
def __init__(self, dbh, domainname, basedir=None, transport=None):
"""Creates a new Domain instance.
@@ -136,7 +139,7 @@
self._prepare()
dbc = self._dbh.cursor()
dbc.execute("INSERT INTO domain_data (gid, tid, domaindir)\
- VALUES (%s, %s, %s)", self._id, self._transport.getID(), self._domaindir)
+ VALUES (%s, %s, %s)", self._id, self._transport.id, self._domaindir)
dbc.execute("INSERT INTO domain_name (domainname, gid, is_primary)\
VALUES (%s, %s, %s)", self._name, self._id, True)
self._dbh.commit()
@@ -155,8 +158,9 @@
if self._id > 0:
self._chkDelete(delUser, delAlias)
dbc = self._dbh.cursor()
- for t in ('alias','users','relocated','domain_name','domain_data'):
- dbc.execute("DELETE FROM %s WHERE gid = %d" % (t, self._id))
+ for tbl in ('alias', 'users', 'relocated', 'domain_name',
+ 'domain_data'):
+ dbc.execute("DELETE FROM %s WHERE gid = %d" % (tbl, self._id))
self._dbh.commit()
dbc.close()
else:
@@ -171,17 +175,17 @@
force -- True/False force new transport for all accounts (bool)
"""
if self._id > 0:
- if transport == self._transport.getTransport():
+ if transport == self._transport.transport:
return
trsp = Transport(self._dbh, transport=transport)
dbc = self._dbh.cursor()
dbc.execute("UPDATE domain_data SET tid = %s WHERE gid = %s",
- trsp.getID(), self._id)
+ trsp.id, self._id)
if dbc.rowcount > 0:
self._dbh.commit()
if force:
dbc.execute("UPDATE users SET tid = %s WHERE gid = %s",
- trsp.getID(), self._id)
+ trsp.id, self._id)
if dbc.rowcount > 0:
self._dbh.commit()
dbc.close()
@@ -199,11 +203,11 @@
def getTransport(self):
"""Returns domain's transport."""
- return self._transport.getTransport()
+ return self._transport.transport
def getTransportID(self):
"""Returns the ID from the domain's transport."""
- return self._transport.getID()
+ return self._transport.id
def getInfo(self):
"""Returns a dictionary with information about the domain."""
@@ -242,7 +246,7 @@
"""Returns a list with all aliases from the domain."""
dbc = self._dbh.cursor()
dbc.execute("SELECT DISTINCT address FROM alias WHERE gid = %s\
- ORDER BY address", self._id)
+ ORDER BY address", self._id)
addresses = dbc.fetchall()
dbc.close()
aliases = []
@@ -312,6 +316,7 @@
del doms
return order, domdict
+
def get_gid(dbh, domainname):
"""Returns the group id of the domain *domainname*.
@@ -326,3 +331,6 @@
if gid:
return gid[0]
return 0
+
+
+del _
--- a/VirtualMailManager/Transport.py Mon Mar 01 04:46:46 2010 +0000
+++ b/VirtualMailManager/Transport.py Mon Mar 01 05:31:43 2010 +0000
@@ -2,89 +2,95 @@
# Copyright (c) 2008 - 2010, Pascal Volk
# See COPYING for distribution information.
-"""Virtual Mail Manager's Transport class to manage the transport for
-domains and accounts."""
+"""
+ VirtualMailManager.Transport
-import VirtualMailManager.constants.ERROR as ERR
+ Virtual Mail Manager's Transport class to manage the transport for
+ domains and accounts.
+"""
+
+from VirtualMailManager.constants.ERROR import UNKNOWN_TRANSPORT_ID
from VirtualMailManager.errors import TransportError
+from VirtualMailManager.pycompat import any
+
class Transport(object):
"""A wrapper class that provides access to the transport table"""
- __slots__ = ('__id', '__transport', '_dbh')
+ __slots__ = ('_id', '_transport', '_dbh')
+
def __init__(self, dbh, tid=None, transport=None):
"""Creates a new Transport instance.
- Either tid or transport must be specified.
+ Either tid or transport must be specified. When both arguments
+ are given, tid will be used.
Keyword arguments:
dbh -- a pyPgSQL.PgSQL.connection
- tid -- the id of a transport (long)
+ tid -- the id of a transport (int/long)
transport -- the value of the transport (str)
+
"""
self._dbh = dbh
- if tid is None and transport is None:
- raise TransportError(
- _('Either tid or transport must be specified.'),
- ERR.TRANSPORT_INIT)
- elif tid is not None:
- try:
- self.__id = long(tid)
- except ValueError:
- raise TransportError(_('tid must be an int/long.'),
- ERR.TRANSPORT_INIT)
+ assert any((tid, transport))
+ if tid:
+ assert not isinstance(tid, bool) and isinstance(tid, (int, long))
+ self._id = tid
self._loadByID()
else:
- self.__transport = transport
+ assert isinstance(transport, basestring)
+ self._transport = transport
self._loadByName()
+ @property
+ def id(self):
+ """The transport's unique ID."""
+ return self._id
+
+ @property
+ def transport(self):
+ """The transport's value, ex: 'dovecot:'"""
+ return self._transport
+
def __eq__(self, other):
if isinstance(other, self.__class__):
- return self.__id == other.getID()
+ return self._id == other.id
return NotImplemented
def __ne__(self, other):
if isinstance(other, self.__class__):
- return self.__id != other.getID()
+ return self._id != other.id
return NotImplemented
def __str__(self):
- return self.__transport
+ return self._transport
def _loadByID(self):
dbc = self._dbh.cursor()
- dbc.execute('SELECT transport FROM transport WHERE tid = %s', self.__id)
+ dbc.execute('SELECT transport FROM transport WHERE tid = %s', self._id)
result = dbc.fetchone()
dbc.close()
- if result is not None:
- self.__transport = result[0]
+ if result:
+ self._transport = result[0]
else:
raise TransportError(_('Unknown tid specified.'),
- ERR.UNKNOWN_TRANSPORT_ID)
+ UNKNOWN_TRANSPORT_ID)
def _loadByName(self):
dbc = self._dbh.cursor()
dbc.execute('SELECT tid FROM transport WHERE transport = %s',
- self.__transport)
+ self._transport)
result = dbc.fetchone()
dbc.close()
- if result is not None:
- self.__id = result[0]
+ if result:
+ self._id = result[0]
else:
self._save()
def _save(self):
dbc = self._dbh.cursor()
dbc.execute("SELECT nextval('transport_id')")
- self.__id = dbc.fetchone()[0]
- dbc.execute('INSERT INTO transport (tid, transport) VALUES (%s, %s)',
- self.__id, self.__transport)
+ self._id = dbc.fetchone()[0]
+ dbc.execute('INSERT INTO transport VALUES (%s, %s)', self._id,
+ self._transport)
self._dbh.commit()
dbc.close()
-
- def getID(self):
- """Returns the unique ID of the transport."""
- return self.__id
-
- def getTransport(self):
- """Returns the value of transport, ex: 'dovecot:'"""
- return self.__transport