VMM/{account,handler}: Account delete, use the boolean keyword force too.
# -*- coding: UTF-8 -*-# Copyright (c) 2007 - 2010, Pascal Volk# See COPYING for distribution information.""" VirtualMailManager.account ~~~~~~~~~~~~~~~~~~~~~~~~~~ Virtual Mail Manager's Account class to manage e-mail accounts."""fromVirtualMailManager.domainimportDomainfromVirtualMailManager.emailaddressimportEmailAddressfromVirtualMailManager.transportimportTransportfromVirtualMailManager.commonimportversion_strfromVirtualMailManager.constantsimport \ACCOUNT_EXISTS,ACCOUNT_MISSING_PASSWORD,ALIAS_PRESENT, \INVALID_ARGUMENT,INVALID_MAIL_LOCATION,NO_SUCH_ACCOUNT, \NO_SUCH_DOMAIN,UNKNOWN_SERVICEfromVirtualMailManager.errorsimportAccountErrorasAErrfromVirtualMailManager.maillocationimportMailLocationfromVirtualMailManager.passwordimportpwhash_=lambdamsg:msgcfg_dget=lambdaoption:NoneclassAccount(object):"""Class to manage e-mail accounts."""__slots__=('_addr','_dbh','_domain','_mail','_new','_passwd','_transport','_uid')def__init__(self,dbh,address):"""Creates a new Account instance. When an account with the given *address* could be found in the database all relevant data will be loaded. Arguments: `dbh` : pyPgSQL.PgSQL.Connection A database connection for the database access. `address` : VirtualMailManager.EmailAddress.EmailAddress The e-mail address of the (new) Account. """ifnotisinstance(address,EmailAddress):raiseTypeError("Argument 'address' is not an EmailAddress")self._addr=addressself._dbh=dbhself._domain=Domain(self._dbh,self._addr.domainname)ifnotself._domain.gid:raiseAErr(_(u"The domain '%s' doesn't exist.")%self._addr.domainname,NO_SUCH_DOMAIN)self._uid=0self._mail=Noneself._transport=self._domain.transportself._passwd=Noneself._new=Trueself._load()def__nonzero__(self):"""Returns `True` if the Account is known, `False` if it's new."""returnnotself._newdef_load(self):"""Load 'uid', 'mid' and 'tid' from the database and set _new to `False` - if the user could be found. """dbc=self._dbh.cursor()dbc.execute('SELECT uid, mid, tid FROM users WHERE gid = %s AND ''local_part = %s',self._domain.gid,self._addr.localpart)result=dbc.fetchone()dbc.close()ifresult:self._uid,_mid,_tid=resultif_tid!=self._transport.tid:self._transport=Transport(self._dbh,tid=_tid)self._mail=MailLocation(self._dbh,mid=_mid)self._new=Falsedef_set_uid(self):"""Set the unique ID for the new Account."""assertself._uid==0dbc=self._dbh.cursor()dbc.execute("SELECT nextval('users_uid')")self._uid=dbc.fetchone()[0]dbc.close()def_prepare(self,maillocation):"""Check and set different attributes - before we store the information in the database. """ifmaillocation.dovecot_version>cfg_dget('misc.dovecot_version'):raiseAErr(_(u"The mailbox format '%(mbfmt)s' requires Dovecot "u">= v%(version)s")%{'mbfmt':maillocation.mbformat,'version':version_str(maillocation.dovecot_version)},INVALID_MAIL_LOCATION)ifnotmaillocation.postfixand \self._transport.transport.lower()in('virtual:','virtual'):raiseAErr(_(u"Invalid transport '%(transport)s' for mailbox "u"format '%(mbfmt)s'")%{'transport':self._transport,'mbfmt':maillocation.mbformat},INVALID_MAIL_LOCATION)self._mail=maillocationself._set_uid()def_switch_state(self,state,service):"""Switch the state of the Account's services on or off. See Account.enable()/Account.disable() for more information."""self._chk_state()ifservicenotin(None,'all','imap','pop3','sieve','smtp'):raiseAErr(_(u"Unknown service: '%s'.")%service,UNKNOWN_SERVICE)ifcfg_dget('misc.dovecot_version')>=0x10200b02:sieve_col='sieve'else:sieve_col='managesieve'ifservicein('smtp','pop3','imap'):sql='UPDATE users SET %s = %s WHERE uid = %d'%(service,state,self._uid)elifservice=='sieve':sql='UPDATE users SET %s = %s WHERE uid = %d'%(sieve_col,state,self._uid)else:sql='UPDATE users SET smtp = %(s)s, pop3 = %(s)s, imap = %(s)s,\%(col)s = %(s)s WHERE uid = %(uid)d'% \{'s':state,'col':sieve_col,'uid':self._uid}dbc=self._dbh.cursor()dbc.execute(sql)ifdbc.rowcount>0:self._dbh.commit()dbc.close()def_count_aliases(self):"""Count all alias addresses where the destination address is the address of the Account."""dbc=self._dbh.cursor()sql="SELECT COUNT(destination) FROM alias WHERE destination = '%s'"\%self._addrdbc.execute(sql)a_count=dbc.fetchone()[0]dbc.close()returna_countdef_chk_state(self):"""Raise an AccountError if the Account is new - not yet saved in the database."""ifself._new:raiseAErr(_(u"The account '%s' doesn't exist.")%self._addr,NO_SUCH_ACCOUNT)@propertydefaddress(self):"""The Account's EmailAddress instance."""returnself._addr@propertydefdomain(self):"""The Domain to which the Account belongs to."""ifself._domain:returnself._domainreturnNone@propertydefgid(self):"""The Account's group ID."""ifself._domain:returnself._domain.gidreturnNone@propertydefhome(self):"""The Account's home directory."""ifnotself._new:return'%s/%s'%(self._domain.directory,self._uid)returnNone@propertydefmail_location(self):"""The Account's MailLocation."""returnself._mail@propertydefuid(self):"""The Account's unique ID."""returnself._uiddefset_password(self,password):"""Set a password for the new Account. If you want to update the password of an existing Account use Account.modify(). Argument: `password` : basestring The password for the new Account. """ifnotisinstance(password,basestring)ornotpassword:raiseAErr(_(u"Couldn't accept password: '%s'")%password,ACCOUNT_MISSING_PASSWORD)self._passwd=passworddefset_transport(self,transport):"""Set the transport for the new Account. If you want to update the transport of an existing Account use Account.modify(). Argument: `transport` : basestring The string representation of the transport, e.g.: 'dovecot:' """self._transport=Transport(self._dbh,transport=transport)defenable(self,service=None):"""Enable a/all service/s for the Account. Possible values for the *service* are: 'imap', 'pop3', 'sieve' and 'smtp'. When all services should be enabled, use 'all' or the default value `None`. Arguments: `service` : basestring The name of a service ('imap', 'pop3', 'smtp', 'sieve'), 'all' or `None`. """self._switch_state(True,service)defdisable(self,service=None):"""Disable a/all service/s for the Account. For more information see: Account.enable()."""self._switch_state(False,service)defsave(self):"""Save the new Account in the database."""ifnotself._new:raiseAErr(_(u"The account '%s' already exists.")%self._addr,ACCOUNT_EXISTS)ifnotself._passwd:raiseAErr(_(u"No password set for '%s'.")%self._addr,ACCOUNT_MISSING_PASSWORD)ifcfg_dget('misc.dovecot_version')>=0x10200b02:sieve_col='sieve'else:sieve_col='managesieve'self._prepare(MailLocation(self._dbh,mbfmt=cfg_dget('mailbox.format'),directory=cfg_dget('mailbox.root')))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,pwhash(self._passwd,user=self._addr),self._uid,self._domain.gid,self._mail.mid,self._transport.tid,cfg_dget('account.smtp'),cfg_dget('account.pop3'),cfg_dget('account.imap'),cfg_dget('account.sieve'))dbc=self._dbh.cursor()dbc.execute(sql)self._dbh.commit()dbc.close()self._new=Falsedefmodify(self,field,value):"""Update the Account's *field* to the new *value*. Possible values for *field* are: 'name', 'password' and 'transport'. *value* is the *field*'s new value. Arguments: `field` : basestring The attribute name: 'name', 'password' or 'transport' `value` : basestring The new value of the attribute. """iffieldnotin('name','password','transport'):raiseAErr(_(u"Unknown field: '%s'")%field,INVALID_ARGUMENT)self._chk_state()dbc=self._dbh.cursor()iffield=='password':dbc.execute('UPDATE users SET passwd = %s WHERE uid = %s',pwhash(value,user=self._addr),self._uid)eliffield=='transport':ifvalue!=self._transport.transport:self._transport=Transport(self._dbh,transport=value)dbc.execute('UPDATE users SET tid = %s WHERE uid = %s',self._transport.tid,self._uid)else:dbc.execute('UPDATE users SET name = %s WHERE uid = %s',value,self._uid)ifdbc.rowcount>0:self._dbh.commit()dbc.close()defget_info(self):"""Returns a dict with some information about the Account. The keys of the dict are: 'address', 'gid', 'home', 'imap' 'mail_location', 'name', 'pop3', 'sieve', 'smtp', transport' and 'uid'. """self._chk_state()ifcfg_dget('misc.dovecot_version')>=0x10200b02:sieve_col='sieve'else:sieve_col='managesieve'sql='SELECT name, smtp, pop3, imap, %s FROM users WHERE uid = %d'% \(sieve_col,self._uid)dbc=self._dbh.cursor()dbc.execute(sql)info=dbc.fetchone()dbc.close()ifinfo:keys=('name','smtp','pop3','imap',sieve_col)info=dict(zip(keys,info))forserviceinkeys[1:]:ifinfo[service]:# TP: A service (pop3/imap) is enabled/usable for a userinfo[service]=_('enabled')else:# TP: A service (pop3/imap) isn't enabled/usable for a userinfo[service]=_('disabled')info['address']=self._addrinfo['gid']=self._domain.gidinfo['home']='%s/%s'%(self._domain.directory,self._uid)info['mail_location']=self._mail.mail_locationinfo['transport']=self._transport.transportinfo['uid']=self._uidreturninfo# nearly impossible‽raiseAErr(_(u"Couldn't fetch information for account: '%s'")%self._addr,NO_SUCH_ACCOUNT)defget_aliases(self):"""Return a list with all alias e-mail addresses, whose destination is the address of the Account."""self._chk_state()dbc=self._dbh.cursor()dbc.execute("SELECT address ||'@'|| domainname FROM alias, ""domain_name WHERE destination = %s AND domain_name.gid = ""alias.gid AND domain_name.is_primary ORDER BY address",str(self._addr))addresses=dbc.fetchall()dbc.close()aliases=[]ifaddresses:aliases=[alias[0]foraliasinaddresses]returnaliasesdefdelete(self,force=False):"""Delete the Account from the database. Argument: `force` : bool if *force* is `True`, all aliases, which points to the Account, will be also deleted. If there are aliases and *force* is `False`, an AccountError will be raised. """ifnotisinstance(force,bool):raiseTypeError('force must be a bool')self._chk_state()dbc=self._dbh.cursor()ifforce:dbc.execute('DELETE FROM users WHERE uid = %s',self._uid)# delete also all aliases where the destination address is the same# as for this account.dbc.execute("DELETE FROM alias WHERE destination = %s",str(self._addr))self._dbh.commit()else:# check first for aliasesa_count=self._count_aliases()ifa_count>0:dbc.close()raiseAErr(_(u"There are %(count)d aliases with the "u"destination address '%(address)s'.")%{'count':a_count,'address':self._addr},ALIAS_PRESENT)dbc.execute('DELETE FROM users WHERE uid = %s',self._uid)self._dbh.commit()dbc.close()self._new=Trueself._uid=0self._addr=self._dbh=self._domain=self._passwd=Noneself._mail=self._transport=Nonedefget_account_by_uid(uid,dbh):"""Search an Account by its UID. This function returns a dict (keys: 'address', 'gid' and 'uid'), if an Account with the given *uid* exists. Argument: `uid` : long The Account unique ID. `dbh` : pyPgSQL.PgSQL.Connection a database connection for the database access. """try:uid=long(uid)exceptValueError:raiseAErr(_(u'UID must be an int/long.'),INVALID_ARGUMENT)ifuid<1:raiseAErr(_(u'UID must be greater than 0.'),INVALID_ARGUMENT)dbc=dbh.cursor()dbc.execute("SELECT local_part||'@'|| domain_name.domainname AS address, ""uid, users.gid FROM users LEFT JOIN domain_name ON ""(domain_name.gid = users.gid AND is_primary) WHERE uid = %s",uid)info=dbc.fetchone()dbc.close()ifnotinfo:raiseAErr(_(u"There is no account with the UID '%d'.")%uid,NO_SUCH_ACCOUNT)info=dict(zip(('address','uid','gid'),info))returninfodel_,cfg_dget