Update of /cvsroot/mailman/mailman3/src/mailman/berkeleydb
In directory sc8-pr-cvs1.sourceforge.net:/tmp/cvs-serv31133/src/mailman/berkeleydb
Added Files:
Tag: 3.0
__init__.py bdblist.py bdbuser.py berkeleydb.py txnprop.py
Log Message:
Initial checkin of Mailman3 source
--- NEW FILE: __init__.py ---
--- NEW FILE: bdblist.py ---
# Copyright (C) 2004 Barry A. Warsaw
#
# This program is free software; you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free
# Software Foundation; either version 2 of the License, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 59 Temple
# Place - Suite 330, Boston, MA 02111-1307, USA.
"""Database for all list configurations."""
import cPickle
import logging
import datetime
from sets import Set
from mailman import errors
from mailman.berkeleydb.berkeleydb import db, BerkeleyDB
from mailman.berkeleydb.txnprop import txnprop
from mailman.interfaces.storages import IListStorage
log = logging.getLogger('mailman.database')
_missing = object()
class BDBListDatabase(BerkeleyDB):
# DATA ABOUT THE DATABASE
#
# metainfo -- {key -> value}
# This table contains storage metadata information. The keys and
# values are simple strings of variable length. Here are the valid
# keys:
#
# version - the version of the database
#
# lists -- {listid -- creation date}
# This table maps from list ids to the date of the list's creation.
# It can also be used to find out whether a list exists or not, as
# every list must have an entry in this table.
#
# variables -- {varname.listid -> value}
# This contains all the configuration options for all the mailing
# lists. It maps from a fully qualified variable name
# (i.e. varname.listid) to the pickled value of the variable.
__implements__ = BerkeleyDB.__implements__ + (IListStorage,)
STORAGE_VERSION = 'ML01'
def _init(self):
self._lists = self._mktable('lists')
self._variables = self._mktable('variables')
# IListDatabase
def _set(self, listid, varname, data):
# We tack the listid onto the front of the varname so that it's easier
# to find all the variables for a particular list, using BerkeleyDB's
# BTree range searching.
fqdnvar = (listid + '.' + varname).encode('utf-8')
pck = cPickle.dumps(data, cPickle.HIGHEST_PROTOCOL)
self._variables.put(fqdnvar, pck, txn=self.txn)
set = txnprop(_set)
def get(self, listid, varname):
val = getattr(self, varname, _missing)
if val is not _missing:
return val
# We tack the listid onto the front of the varname so that it's easier
# to find all the variables for a particular list, using BerkeleyDB's
# BTree range searching.
fqdnvar = (listid + '.' + varname).encode('utf-8')
data = self._variables.get(fqdnvar, _missing, txn=self.txn)
if data is _missing:
raise AttributeError(fqdnvar)
return cPickle.loads(data)
def list_exists(self, listid):
listid = listid.encode('utf-8')
return self._lists.get(listid, txn=self.txn) is not None
def _create_list(self, listid):
if self.list_exists(listid):
raise errors.ListExistsError(listid)
utcnow = datetime.datetime.utcnow()
data = cPickle.dumps(utcnow, cPickle.HIGHEST_PROTOCOL)
self._lists.put(listid.encode('utf-8'), data, txn=self.txn)
return utcnow
create_list = txnprop(_create_list)
def _remove_list(self, listid):
if not self.list_exists(listid):
raise errors.NoListError(listid)
# Find and remove all variables qualified by the listid
c = self._variables.cursor(txn=self.txn)
try:
try:
rec = c.set_range(listid)
except db.DBNotFoundError:
log.error('List exists, but has no variables? %s', listid)
rec = None
prefix = listid + '.'
while rec:
key, val = rec
if not key.startswith(prefix):
break
c.delete()
rec = c.next()
finally:
c.close()
# Remove the entry in the lists table.
self._lists.delete(listid, txn=self.txn)
remove_list = txnprop(_remove_list)
def listids(self):
# BAW: This can't be a generator because of the finally :/
listids = []
c = self._lists.cursor(txn=self.txn)
try:
rec = c.first()
while rec:
listids.append(rec[0])
rec = c.next()
finally:
c.close()
return listids
def domains(self):
domains = Set()
for listid in self.listids():
listname, domain = listid.split('@', 1)
domains.add(domain)
return list(s)
--- NEW FILE: bdbuser.py ---
# Copyright (C) 2004 Barry A. Warsaw
#
# This program is free software; you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free
# Software Foundation; either version 2 of the License, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 59 Temple
# Place - Suite 330, Boston, MA 02111-1307, USA.
"""Database for list rosters."""
import os
import sha
import struct
import cPickle
import datetime
from mailman import errors
from mailman import lockfile
from mailman.berkeleydb.berkeleydb import db, BerkeleyDB
from mailman.berkeleydb.txnprop import txnprop
from mailman.config import configuration
from mailman.dbmgr import dbmgr
from mailman.interfaces.storages import IUserStorage
from mailman.interfaces.user import IUser
_missing = object()
def uid2int(s):
return struct.unpack('>Q', s)[0]
def int2uid(i):
return struct.pack('>Q', i)
class BDBUserDatabase(BerkeleyDB):
# DATA ABOUT THE DATABASE
#
# metainfo -- {key -> value}
# This table contains storage metadata information. The keys and
# values are simple strings of variable length. Here are the
# valid keys:
#
# version - the version of the database
# nextuid - the next available user id
#
# GLOBAL MEMBER DATA
#
# users -- {userid -> createtime}
# Maps unique user ids to the timestamp that the user was created.
#
# passwords -- {userid -> password}
# Maps user ids to that person's hashed passwords.
#
# addresses -- {userid\0label -> address}
# Maps userid/label keys to email addresses
#
# realnames -- {userid\0label -> realname}
# Maps userid/label keys to real names
__implements__ = BerkeleyDB.__implements__ + (IUserStorage,)
STORAGE_VERSION = 'USR01'
def _init(self):
self._users = self._mktable('users')
self._passwords = self._mktable('passwords')
self._addresses = self._mktable('addresses')
self._realnames = self._mktable('realnames')
self._uidlock = lockfile.LockFile(
os.path.join(configuration.layout.lockdir, 'uid.lock'))
self._uidlock.lock()
try:
nextuid = self._metainfo.get('nextuid')
if nextuid is None:
self._inituids()
finally:
self._uidlock.unlock()
def _inituids(self):
# BAW: reserve uid 0 for 'something special'
self._metainfo.put('nextuid', pack('>Q', 1), txn=self.txn)
_inituids = txnprop(_inituids)
# IUserStorage
def _get_user(self, uid, domain):
# BAW: we ignore the domain because that's only useful when there are
# multiple backends.
#
# The keys are always utf-8 encoded because the bsddb module does not
# allow unicode keys.
suid = int2uid(uid)
if self._users.get(suid, _missing, txn=self.txn) is _missing:
raise errors.NoUserError(uid)
return User(self, suid)
get_user = txnprop(_get_user)
def _create_user(self, domain):
# We need to calculate a new user id that's guaranteed to be
# unique. To do this, we'll acquire a global file lock.
self._uidlock.lock()
try:
uid = self._metainfo.get('nextuid', txn=self.txn)
next = pack('>Q', unpack('>Q', uid)[0] + 1)
self._metainfo.put('nextuid', next, txn=self.txn)
finally:
self._uidlock.unlock()
utcnow = datetime.datetime.utcnow()
data = cPickle.dumps(utcnow, cPickle.HIGHEST_PROTOCOL)
self._users.put(uid, data, txn=self.txn)
return User(self, uid)
create_user = txnprop(_create_user)
# User class helpers
def _get_password(self, uid):
return self._passwords.get(uid, txn=self.txn)
get_password = txnprop(_get_password)
def _set_password(self, uid, password):
self._passwords.put(uid, password, txn=self.txn)
set_password = txnprop(_set_password)
def _get_address(self, uid, label):
# labels may be unicodes
enc = label.encode('utf-8')
key = '%s\0%s' % (uid, enc)
return self._addresses.get(key, txn=self.txn)
get_address = txnprop(_get_address)
def _add_address(self, uid, label, address):
enc = label.encode('utf-8')
key = '%s\0%s' % (uid, enc)
self._addresses.put(key, address, txn=self.txn)
add_address = txnprop(_add_address)
def _delete_address(self, uid, label):
enc = label.encode('utf-8')
key = '%s\0%s' % (uid, enc)
self._addresses.delete(key, txn=self.txn)
delete_address = txnprop(_delete_address)
def _get_realname(self, uid, label):
# labels may be unicodes
enc = label.encode('utf-8')
key = '%s\0%s' % (uid, enc)
return self._realnames.get(key, txn=self.txn)
get_realname = txnprop(_get_realname)
def _set_realname(self, uid, label, realname):
enc = label.encode('utf-8')
key = '%s\0%s' % (uid, enc)
self._realnames.put(key, realname, txn=self.txn)
set_realname = txnprop(_set_realname)
def _get_labels(self, uid):
# BAW: this can't be a generator because of the finally ;/
labels = []
c = self._addresses.cursor(txn=self.txn)
try:
try:
rec = c.set_range(uid)
except db.DBNotFoundError:
rec = None
while rec:
key, val = rec
rec = c.next()
kuid, label = key.split('\0', 1)
if kuid <> uid:
break
labels.append(label)
finally:
c.close()
return labels
class readonly(object):
def __init__(self, getf):
self._getf = getf
def __get__(self, obj, type=None):
return self._getf(obj)
def __set__(self, obj, val):
raise AttributeError(obj)
def __delete__(self, obj):
raise AttributeError(obj)
class User(object):
__implements__ = IUser
def __init__(self, db, uid):
self._db = db
self._uid = uid
def _get_uid(self):
return uid2int(self._uid)
uid = readonly(_get_uid)
def _get_password(self):
return self._db.get_password(self._uid)
def _set_password(self, password):
hashed = sha.new(password).digest()
self._db.set_password(self._uid, hashed)
password = property(_get_password, _set_password)
def authenticate(self, challenge):
hashed = sha.new(challenge).digest()
return hashed == self.password
def get_address(self, label):
if self._db.get_address(self._uid, label):
return Address(self._db, self._uid, label)
return None
def add_address(self, label, address):
if self._db.get_address(self._uid, label):
raise UserAddressError(label)
self._db.add_address(self._uid, label, address)
return Address(self._db, self._uid, label)
def delete_address(self, label):
if not self._db.get_address(self._uid, label):
raise UserAddressError(label)
self._db.delete_address(self._uid, label)
def _addresses(self):
for label in self._db.get_labels(self._uid):
yield Address(self._db, self._uid, label)
address = readonly(_addresses)
class Address(object):
def __init__(self, db, uid, label):
self._db = db
self._uid = uid
self._label = label
def _get_user(self):
return User(self._db, self._uid)
user = readonly(_get_user)
def _get_label(self):
return self._label
label = readonly(_get_label)
def _get_address(self):
return self._db.get_address(self._uid, self._label)
def _set_address(self, address):
self._db.add_address(self._uid, self._label, address)
address = property(_get_address, _set_address)
def _get_realname(self):
return self._db.get_realname(self._uid, self._label)
def _set_realname(self, realname):
self._db.set_realname(self._uid, self._label, realname)
realname = property(_get_realname, _set_realname)
--- NEW FILE: berkeleydb.py ---
# Copyright (C) 2004 Barry A. Warsaw
#
# This program is free software; you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free
# Software Foundation; either version 2 of the License, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 59 Temple
# Place - Suite 330, Boston, MA 02111-1307, USA.
import os
import errno
import logging
import datetime
import threading
from bsddb import db
from mailman import errors
from mailman import lockfile
from mailman.berkeleydb.txnprop import txnprop
from mailman.config import configuration
from mailman.interfaces.storages import IStorage
from mailman.interfaces.transactions import ITransaction, ITransactionManager
from mailman.workthread import WorkThread
GBYTES = 1024 * 1024 * 1000
log = logging.getLogger('mailman.database')
class BerkeleyDB(object):
# DATA ABOUT THE DATABASE
#
# metainfo -- {key -> value}
# This table contains storage metadata information. The keys and
# values are simple strings of variable length. Here are the valid
# keys:
#
# version - the version of the database
__implements__ = ITransactionManager, IStorage
def __init__(self, path, checkpoint_interval, cachesize):
self._path = path
self._cpinterval = checkpoint_interval
self._cachesize = cachesize
self._jointime = checkpoint_interval * 2
self._env = None
def open(self):
if self._env:
raise errors.DatabaseAlreadyOpen
log.info('initializing database: %s', self._path)
try:
os.mkdir(self._path, 02775)
except OSError, e:
if e.errno <> errno.EEXIST: raise
# Create the environment object
self._env = env = db.DBEnv()
# Acquire the lock file to serialize the creation of the lockfile.
#import pdb ; pdb.set_trace()
lockpath = os.path.join(configuration.layout.lockdir, 'bdb.lock')
lock = lockfile.LockFile(lockpath)
status = lock.lock()
if status in (lockfile.WasStale, lockfile.WasBroken):
# The other process died while holding the lock or held it too long
recoverflag = db.DB_RECOVER_FATAL
else:
recoverflag = db.DB_RECOVER
# If an environment exists, we must join it rather than open it,
# otherwise we'll get nasty DB_RUNRECOVERY errors when the second
# process to open the environment begins a transaction.
try:
# We must join an existing environment, otherwise we'll get
# DB_RUNRECOVERY errors when the second process to open the
# environment begins a transaction.
log.debug('joining environment: %s', self._path)
env.open(self._path, db.DB_JOINENV)
except db.DBNoSuchFileError, e:
if e[0] <> errno.ENOENT: raise
# So the environment didn't exist -- create it.
log.debug('creating environment: %s', self._path)
# Set the BerkeleyDB cache size
gbytes, bytes = divmod(self._cachesize, GBYTES)
env.set_cachesize(gbytes, bytes)
env.open(self._path,
db.DB_CREATE |
db.DB_THREAD |
db.DB_INIT_LOCK |
db.DB_INIT_LOG |
db.DB_INIT_MPOOL |
db.DB_INIT_TXN |
recoverflag
)
# Release the creation lock
lock.unlock()
self._tables = []
self._metainfo = self._mktable('metainfo')
self._init()
# Start a checkpointing thread so any potential recovery can be much
# faster. Right now, we hardcode the checkpointing interval.
self._chkstop = threading.Event()
self._chkpointer = Checkpointer(
self, self._chkstop, self._cpinterval)
self._chkpointer.start()
# Check the database version number
version = self._metainfo.get('version')
if version is None:
self._setversion(self.STORAGE_VERSION)
else:
# Currently there's nothing to upgrade
assert version == self.STORAGE_VERSION
log.info('database ready: %s', self._path)
def close(self):
# Abort any in-progress transaction
t = threading.currentThread()
try:
t.txnstack.reverse()
for txn in t.txnstack:
txn.abort()
t.txnstack = []
except AttributeError:
pass
for d in self._tables:
d.close()
# Turn off the checkpointing thread.
self._chkstop.set()
self._chkpointer.join(self._jointime)
# Checkpoint the database twice, as recommended by Sleepycat
self.checkpoint()
self.checkpoint()
self._env.close()
log.info('database closed: %s', self._path)
def _setversion(self, version):
self._metainfo.put('version', version, txn=self.txn)
_setversion = txnprop(_setversion)
def _init(self):
raise NotImplementedError
def _mktable(self, name, flags=0):
d = db.DB(self._env)
if flags:
d.set_flags(flags)
openflags = db.DB_CREATE
# db 4.1 requires that databases be opened in a transaction. We'll
# use auto commit, but only if that flag exists (i.e. we're using at
# least db 4.1).
try:
openflags |= db.DB_AUTO_COMMIT
except AttributeError:
pass
d.open(name, db.DB_BTREE, openflags)
self._tables.append(d)
return d
def checkpoint(self):
log.debug('checkpointing')
self._env.txn_checkpoint(0, 0, db.DB_FORCE)
def _withtxn(self, meth, *args, **kws):
txn = self.begin()
try:
rtn = meth(*args, **kws)
except:
txn.abort()
raise
else:
txn.commit()
return rtn
# ITransactionManager
def _txn(self):
try:
return self._txnstack()[-1]
except (AttributeError, IndexError):
# The only way we get here is if we're doing a get operation
# outside of any explicit transaction. I think we're safe simply
# returning None and letting BerkeleyDB do the access outside of
# the transaction. I'm not positive about this, and if we have
# problems (like deadlocks), the answer will be to wrap all get
# methods in _withtxn().
return None
txn = property(_txn)
def _txnstack(self):
t = threading.currentThread()
return t.dbtxns[self.__class__.__name__]
def begin(self):
t = threading.currentThread()
try:
dbtxns = t.dbtxns
except AttributeError:
dbtxns = t.dbtxns = {}
stack = dbtxns.setdefault(self.__class__.__name__, [])
if stack:
txn = self._env.txn_begin(parent=stack[-1])
else:
txn = self._env.txn_begin()
stack.append(txn)
return TxnWrapper(self, txn)
def poptxn(self):
self._txnstack().pop()
class Checkpointer(WorkThread):
def __init__(self, db, event, checkpoint_interval):
WorkThread.__init__(self, event, log, checkpoint_interval)
self._db = db
def _work(self):
self._db.checkpoint()
class TxnWrapper(object):
__implements__ = ITransaction
def __init__(self, db, txn):
self._db = db
self._txn = txn
def commit(self):
self._txn.commit()
self._db.poptxn()
def abort(self):
self._txn.abort()
self._db.poptxn()
--- NEW FILE: txnprop.py ---
# Copyright (C) 2004 Barry A. Warsaw
#
# This program is free software; you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free
# Software Foundation; either version 2 of the License, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 59 Temple
# Place - Suite 330, Boston, MA 02111-1307, USA.
"""A transaction wrapping property."""
# A common pattern is to write a non-public method which does the basic logic
# of the operation, along with public method which wraps the first in a
# transaction or a lock acquire/release. This descriptor class makes this
# pattern much more consistent. Use txnprop when you want to both acquire the
# lock and do the operation in a transaction (i.e. for mutator methods).
class txnprop(object):
def __init__(self, func):
# func is going to be a function not a method (even unbound)
self._func = func
def __get__(self, obj, type=None):
# Return a wrapper function that creates a bound method from the
# function, then calls it wrapped in a transaction/lock acquisition.
# This uses a non-public method caled _withtxn()
def wrapper(*args):
return obj._withtxn(self._func.__get__(obj), *args)
return wrapper
_______________________________________________
Mailman-checkins mailing list
[EMAIL PROTECTED]
Unsubscribe: http://mail.python.org/mailman/options/mailman-checkins/archive%40jab.org
