Converted to use dbi.

This commit is contained in:
Jeremy Fincher 2004-08-11 06:17:11 +00:00
parent 77ce14f481
commit 0768ce1c85
3 changed files with 180 additions and 386 deletions

View File

@ -43,6 +43,7 @@ import sets
import random
import itertools
import supybot.dbi as dbi
import supybot.conf as conf
import supybot.ircdb as ircdb
import supybot.utils as utils
@ -53,50 +54,15 @@ import supybot.privmsgs as privmsgs
import supybot.registry as registry
import supybot.callbacks as callbacks
class FunDBDBInterface(object):
def close(self):
pass
def flush(self):
pass
def get(self, channel, type, id):
"""Returns just the text associated with the channel, type, and id."""
raise NotImplementedError
def info(self, channel, type, id):
"""Returns the test and the metadata associated with the
channel, type, and id."""
raise NotImplementedError
def add(self, channel, type, text, by):
raise NotImplementedError
def remove(self, channel, type, id):
raise NotImplementedError
def change(self, channel, type, id, f):
raise NotImplementedError
def random(self, channel, type):
raise NotImplementedError
def size(self, channel, type):
raise NotImplementedError
def search(self, channel, type, p):
"""Returns a list of (id, text) pairs whose text matches predicate p"""
raise NotImplementedError
class FlatfileFunDBDB(FunDBDBInterface):
class FunDBDB(plugins.FlatfileDB):
def serialize(self, v):
return csv.join(map(str, v))
def deserialize(self, s):
return csv.split(s)
class DbiFunDBDB(object):
class FunDBDB(dbi.DB):
class Record(object):
__metaclass__ = dbi.Record
__fields__ = [
'by',
('text', str),
]
def __init__(self):
self.dbs = ircutils.IrcDict()
self.filenames = sets.Set()
@ -121,49 +87,33 @@ class FlatfileFunDBDB(FunDBDBInterface):
return self.dbs[channel][type]
def get(self, channel, type, id):
return self.info(channel, type, id)[1]
def info(self, channel, type, id):
db = self._getDb(channel, type)
return db.getRecord(id)
return db.get(id)
def add(self, channel, type, text, by):
db = self._getDb(channel, type)
return db.addRecord([by, text])
return db.add(db.Record(by=by, text=text))
def remove(self, channel, type, id):
db = self._getDb(channel, type)
db.delRecord(id)
db.remove(id)
def change(self, channel, type, id, f):
db = self._getDb(channel, type)
(by, text) = db.getRecord(id)
db.setRecord(id, [by, f(text)])
record = db.get(id)
record.text = f(record.text)
db.set(id, record)
def random(self, channel, type):
db = self._getDb(channel, type)
t = random.choice(db.records())
if t is not None:
(id, (by, text)) = t
t = (id, text)
return t
return db.random()
def size(self, channel, type):
db = self._getDb(channel, type)
return itertools.ilen(db.records())
def search(self, channel, type, p):
db = self._getDb(channel, type)
L = []
for (id, record) in db.records():
text = record[1]
if p(text):
L.append((id, text))
return L
return itertools.ilen(db)
def FunDBDB():
return FlatfileFunDBDB()
return DbiFunDBDB()
conf.registerPlugin('FunDB')
conf.registerChannelValue(conf.supybot.plugins.FunDB, 'showIds',
@ -316,8 +266,8 @@ class FunDB(callbacks.Privmsg):
if id is None:
return
try:
text = self.db.get(channel, type, id)
irc.reply(text)
x = self.db.get(channel, type, id)
irc.reply(x.text)
except KeyError:
irc.error('There is no %s with that id.' % type)
@ -336,8 +286,8 @@ class FunDB(callbacks.Privmsg):
if id is None:
return
try:
(text, by) = self.db.info(channel, type, id)
reply = '%s #%s: %r; Created by %s.' % (type, id, text, by)
x = self.db.get(channel, type, id)
reply = '%s #%s: %r; Created by %s.' % (type, x.id, x.text, x.by)
irc.reply(reply)
except KeyError:
irc.error('There is no %s with that id.' % type)
@ -365,14 +315,13 @@ class FunDB(callbacks.Privmsg):
nick = privmsgs.getArgs(args)
if not nick:
raise callbacks.ArgumentError
t = self.db.random(channel, 'insult')
if t is None:
insult = self.db.random(channel, 'insult')
if insult is None:
irc.error('There are currently no available insults.')
else:
(id, insult) = t
nick = self._replaceFirstPerson(nick, msg.nick)
insult = '%s: %s' % (nick, insult.replace('$who', nick))
irc.reply(self._formatResponse(insult, id, channel),
s = '%s: %s' % (nick, insult.text.replace('$who', nick))
irc.reply(self._formatResponse(s, insult.id, channel),
prefixName=False)
def lart(self, irc, msg, args):
@ -400,23 +349,21 @@ class FunDB(callbacks.Privmsg):
if id:
try:
lart = self.db.get(channel, 'lart', id)
t = (id, lart)
except KeyError:
irc.error('There is no such lart.')
return
else:
t = self.db.random(channel, 'lart')
if t is None:
lart = self.db.random(channel, 'lart')
if lart is None:
irc.error('There are currently no available larts.')
else:
(id, lart) = t
nick = self._replaceFirstPerson(nick, msg.nick)
reason = self._replaceFirstPerson(reason, msg.nick)
s = lart.replace('$who', nick)
s = lart.text.replace('$who', nick)
if reason:
s = '%s for %s' % (s, reason)
s = s.rstrip('.')
irc.reply(self._formatResponse(s, id, channel), action=True)
irc.reply(self._formatResponse(s, lart.id, channel), action=True)
def praise(self, irc, msg, args):
"""[<channel>] [<id>] <text> [for <reason>]
@ -442,23 +389,21 @@ class FunDB(callbacks.Privmsg):
if id:
try:
praise = self.db.get(channel, 'praise', id)
t = (id, praise)
except KeyError:
irc.error('There is no such praise.')
return
else:
t = self.db.random(channel, 'praise')
if t is None:
praise = self.db.random(channel, 'praise')
if praise is None:
irc.error('There are currently no available praises.')
else:
(id, praise) = t
nick = self._replaceFirstPerson(nick, msg.nick)
reason = self._replaceFirstPerson(reason, msg.nick)
s = praise.replace('$who', nick)
s = praise.text.replace('$who', nick)
if reason:
s = '%s for %s' % (s, reason)
s = s.rstrip('.')
irc.reply(self._formatResponse(s, id, channel), action=True)
irc.reply(self._formatResponse(s, praise.id, channel), action=True)
Class = FunDB

View File

@ -373,136 +373,6 @@ class PeriodicFileDownloader(object):
world.threadsSpawned += 1
class FlatfileDB(object):
def __init__(self, filename, maxSize=10**6):
self.filename = filename
try:
fd = file(self.filename)
strId = fd.readline().rstrip()
self.maxSize = len(strId)
self.currentId = int(strId)
except EnvironmentError, e:
# File couldn't be opened.
self.maxSize = int(math.log10(maxSize))
self.currentId = 0
self._incrementCurrentId()
def serialize(self, record):
raise NotImplementedError
def deserialize(self, s):
raise NotImplementedError
def _canonicalId(self, id):
if id is not None:
return str(id).zfill(self.maxSize)
else:
return '-'*self.maxSize
def _incrementCurrentId(self, fd=None):
fdWasNone = fd is None
if fdWasNone:
fd = file(self.filename, 'a')
fd.seek(0)
self.currentId += 1
fd.write(self._canonicalId(self.currentId))
fd.write('\n')
if fdWasNone:
fd.close()
def _splitLine(self, line):
line = line.rstrip('\r\n')
(strId, strRecord) = line.split(':', 1)
return (strId, strRecord)
def _joinLine(self, id, record):
return '%s:%s\n' % (self._canonicalId(id), self.serialize(record))
def addRecord(self, record):
line = self._joinLine(self.currentId, record)
try:
fd = file(self.filename, 'r+')
fd.seek(0, 2) # End.
fd.write(line)
return self.currentId
finally:
self._incrementCurrentId(fd)
fd.close()
def getRecord(self, id):
strId = self._canonicalId(id)
try:
fd = file(self.filename)
fd.readline() # First line, nextId.
for line in fd:
(lineId, strRecord) = self._splitLine(line)
if lineId == strId:
return self.deserialize(strRecord)
raise KeyError, id
finally:
fd.close()
def setRecord(self, id, record):
strLine = self._joinLine(id, record)
try:
fd = file(self.filename, 'r+')
self.delRecord(id, fd)
fd.seek(0, 2) # End.
fd.write(strLine)
finally:
fd.close()
def delRecord(self, id, fd=None):
fdWasNone = fd is None
strId = self._canonicalId(id)
try:
if fdWasNone:
fd = file(self.filename, 'r+')
fd.seek(0)
fd.readline() # First line, nextId
pos = fd.tell()
line = fd.readline()
while line:
(lineId, strRecord) = self._splitLine(line)
if lineId == strId:
fd.seek(pos)
fd.write(self._canonicalId(None))
fd.seek(pos)
fd.readline() # Same line we just rewrote the id for.
pos = fd.tell()
line = fd.readline()
# We should be at the end.
finally:
if fdWasNone:
fd.close()
def records(self):
fd = file(self.filename)
fd.readline() # First line, nextId.
for line in fd:
(strId, strRecord) = self._splitLine(line)
if not strId.startswith('-'):
yield (int(strId), self.deserialize(strRecord))
fd.close()
def vacuum(self):
infd = file(self.filename)
outfd = utils.transactionalFile(self.filename)
outfd.write(infd.readline()) # First line, nextId.
for line in infd:
if not line.startswith('-'):
outfd.write(line)
infd.close()
outfd.close()
def flush(self):
pass # No-op, we maintain no open files.
def close(self):
self.vacuum() # Should we do this? It should be fine.
_randomnickRe = re.compile(r'\$rand(?:om)?nick', re.I)
_randomdateRe = re.compile(r'\$rand(?:om)?date', re.I)
_randomintRe = re.compile(r'\$rand(?:omint)?', re.I)

View File

@ -33,186 +33,165 @@ from testsupport import *
import supybot.ircdb as ircdb
try:
import sqlite
except ImportError:
sqlite = None
class TestFunDB(ChannelPluginTestCase, PluginDocumentation):
plugins = ('FunDB','User','Utilities')
def setUp(self):
ChannelPluginTestCase.setUp(self)
self.prefix = 't3st!bar@foo.com'
self.nick = 't3st'
self.irc.feedMsg(ircmsgs.privmsg(self.irc.nick,
'register t3st moo',
prefix=self.prefix))
_ = self.irc.takeMsg()
#ircdb.users.getUser('t3st').addCapability('admin')
ircdb.users.getUser('t3st').addCapability('#test.op')
conf.supybot.plugins.FunDB.showIds.setValue(True)
if sqlite is not None:
class TestFunDB(ChannelPluginTestCase, PluginDocumentation):
plugins = ('FunDB','User','Utilities')
def setUp(self):
ChannelPluginTestCase.setUp(self)
self.prefix = 't3st!bar@foo.com'
self.nick = 't3st'
self.irc.feedMsg(ircmsgs.privmsg(self.irc.nick,
'register t3st moo',
prefix=self.prefix))
_ = self.irc.takeMsg()
#ircdb.users.getUser('t3st').addCapability('admin')
ircdb.users.getUser('t3st').addCapability('#test.op')
conf.supybot.plugins.FunDB.showIds.setValue(True)
def testAdd(self):
self.assertError('add l4rt foo')
self.assertError('add lart foo')
def testAdd(self):
self.assertError('add l4rt foo')
self.assertError('add lart foo')
def testRemove(self):
self.assertError('remove l4rt foo')
self.assertError('remove lart foo')
def testRemove(self):
self.assertError('remove l4rt foo')
self.assertError('remove lart foo')
def testLart(self):
self.assertNotError('add lart jabs $who')
self.assertHelp('lart')
self.assertAction('lart jemfinch for being dumb',
'jabs jemfinch for being dumb (#1)')
self.assertAction('lart jemfinch', 'jabs jemfinch (#1)')
self.assertRegexp('stats lart', 'currently 1 lart')
self.assertNotError('add lart shoots $who')
self.assertHelp('lart 1')
self.assertAction('lart 1 jemfinch', 'jabs jemfinch (#1)')
self.assertAction('lart 2 jemfinch for being dumb',
'shoots jemfinch for being dumb (#2)')
self.assertNotRegexp('lart %s' % self.irc.nick, self.irc.nick)
self.assertNotError('remove lart 1')
self.assertRegexp('stats lart', 'currently 1 lart')
self.assertAction('lart jemfinch', 'shoots jemfinch (#2)')
self.assertNotError('remove lart 2')
self.assertRegexp('stats lart', 'currently 0')
self.assertError('lart jemfinch')
def testLart(self):
self.assertNotError('add lart jabs $who')
self.assertHelp('lart')
self.assertResponse('lart jemfinch for being dumb',
'\x01ACTION jabs jemfinch for being dumb '
'(#1)\x01')
self.assertResponse('lart jemfinch',
'\x01ACTION jabs jemfinch (#1)\x01')
self.assertRegexp('stats lart', 'currently 1 lart')
self.assertNotError('add lart shoots $who')
self.assertHelp('lart 1')
self.assertResponse('lart 1 jemfinch',
'\x01ACTION jabs jemfinch (#1)\x01')
self.assertResponse('lart 2 jemfinch for being dumb',
'\x01ACTION shoots jemfinch for being dumb '
'(#2)\x01')
self.assertNotRegexp('lart %s' % self.irc.nick, self.irc.nick)
self.assertNotError('remove lart 1')
self.assertRegexp('stats lart', 'currently 1 lart')
self.assertResponse('lart jemfinch',
'\x01ACTION shoots jemfinch (#2)\x01')
self.assertNotError('remove lart 2')
self.assertRegexp('stats lart', 'currently 0')
self.assertError('lart jemfinch')
def testLartAndPraiseRemoveTrailingPeriods(self):
for s in ['lart', 'praise']:
self.assertNotError('add %s $who foo!' % s)
self.assertAction('%s bar.' % s, 'bar foo! (#1)')
def testLartAndPraiseRemoveTrailingPeriods(self):
for s in ['lart', 'praise']:
self.assertNotError('add %s $who foo!' % s)
self.assertAction('%s bar.' % s, 'bar foo! (#1)')
def testMyMeReplacement(self):
self.assertNotError('add lart jabs $who')
self.assertNotError('add praise pets $who')
self.assertNotError('add insult foo')
self.assertAction('lart me', 'jabs t3st (#1)')
self.assertAction('praise me', 'pets t3st (#1)')
#self.assertResponse('insult me', 't3st: foo (#1)')
self.assertAction('lart whamme', 'jabs whamme (#1)')
self.assertAction('praise whamme', 'pets whamme (#1)')
#self.assertResponse('insult whamme', 'whamme: foo (#1)')
self.assertAction('lart my knee', 'jabs t3st\'s knee (#1)')
self.assertAction('praise my knee', 'pets t3st\'s knee (#1)')
#self.assertResponse('insult my knee', 't3st\'s knee: foo (#1)')
self.assertAction('lart sammy the snake',
'jabs sammy the snake (#1)')
self.assertAction('praise sammy the snake',
'pets sammy the snake (#1)')
#self.assertResponse('insult sammy the snake',
# 'sammy the snake: foo (#1)')
self.assertAction('lart me for my',
'jabs t3st for t3st\'s (#1)')
self.assertAction('praise me for my',
'pets t3st for t3st\'s (#1)')
self.assertAction('lart me and %s' % self.irc.nick,
'jabs t3st and %s (#1)' % self.irc.nick)
self.assertAction('praise me and %s' % self.irc.nick,
'pets t3st and %s (#1)' % self.irc.nick)
self.assertNotError('remove lart 1')
self.assertNotError('remove praise 1')
self.assertNotError('remove insult 1')
def testMyMeReplacement(self):
self.assertNotError('add lart jabs $who')
self.assertNotError('add praise pets $who')
self.assertNotError('add insult foo')
self.assertAction('lart me', 'jabs t3st (#1)')
self.assertAction('praise me', 'pets t3st (#1)')
#self.assertResponse('insult me', 't3st: foo (#1)')
self.assertAction('lart whamme', 'jabs whamme (#1)')
self.assertAction('praise whamme', 'pets whamme (#1)')
#self.assertResponse('insult whamme', 'whamme: foo (#1)')
self.assertAction('lart my knee', 'jabs t3st\'s knee (#1)')
self.assertAction('praise my knee', 'pets t3st\'s knee (#1)')
#self.assertResponse('insult my knee', 't3st\'s knee: foo (#1)')
self.assertAction('lart sammy the snake',
'jabs sammy the snake (#1)')
self.assertAction('praise sammy the snake',
'pets sammy the snake (#1)')
#self.assertResponse('insult sammy the snake',
# 'sammy the snake: foo (#1)')
self.assertAction('lart me for my',
'jabs t3st for t3st\'s (#1)')
self.assertAction('praise me for my',
'pets t3st for t3st\'s (#1)')
self.assertAction('lart me and %s' % self.irc.nick,
'jabs t3st and %s (#1)' % self.irc.nick)
self.assertAction('praise me and %s' % self.irc.nick,
'pets t3st and %s (#1)' % self.irc.nick)
self.assertNotError('remove lart 1')
self.assertNotError('remove praise 1')
self.assertNotError('remove insult 1')
def testInsult(self):
self.assertNotError('add insult Fatty McFatty')
self.assertResponse('insult jemfinch',
'jemfinch: Fatty McFatty (#1)')
self.assertRegexp('stats insult', r'currently 1')
self.assertNotError('remove insult 1')
self.assertRegexp('stats insult', 'currently 0')
self.assertError('insult jemfinch')
def testInsult(self):
self.assertNotError('add insult Fatty McFatty')
self.assertResponse('insult jemfinch',
'jemfinch: Fatty McFatty (#1)')
self.assertRegexp('stats insult', r'currently 1')
self.assertNotError('remove insult 1')
self.assertRegexp('stats insult', 'currently 0')
self.assertError('insult jemfinch')
def testChannelReplies(self):
self.assertNotError('add #tester praise pets $who')
self.assertNotError('add praise pats $who')
self.assertNotError('add #tester lart stabs $who')
self.assertNotError('add lart stubs $who')
self.assertNotError('add #tester insult nimrod')
self.assertNotError('add insult nimwit')
self.assertAction('praise jemfinch', 'pats jemfinch (#1)')
self.assertAction('praise #tester jemfinch', 'pets jemfinch (#1)')
self.assertAction('lart jemfinch', 'stubs jemfinch (#1)')
self.assertAction('lart #tester jemfinch', 'stabs jemfinch (#1)')
self.assertResponse('insult jemfinch', 'jemfinch: nimwit (#1)')
self.assertResponse('insult #tester jemfinch',
'jemfinch: nimrod (#1)')
def testChannelReplies(self):
self.assertNotError('add #tester praise pets $who')
self.assertNotError('add praise pats $who')
self.assertNotError('add #tester lart stabs $who')
self.assertNotError('add lart stubs $who')
self.assertNotError('add #tester insult nimrod')
self.assertNotError('add insult nimwit')
self.assertResponse('praise jemfinch',
'\x01ACTION pats jemfinch (#1)\x01')
self.assertResponse('praise #tester jemfinch',
'\x01ACTION pets jemfinch (#1)\x01')
self.assertResponse('lart jemfinch',
'\x01ACTION stubs jemfinch (#1)\x01')
self.assertResponse('lart #tester jemfinch',
'\x01ACTION stabs jemfinch (#1)\x01')
self.assertResponse('insult jemfinch', 'jemfinch: nimwit (#1)')
self.assertResponse('insult #tester jemfinch',
'jemfinch: nimrod (#1)')
def testPraise(self):
self.assertNotError('add praise pets $who')
self.assertHelp('praise')
self.assertAction('praise jemfinch for being him',
'pets jemfinch for being him (#1)')
self.assertAction('praise jemfinch', 'pets jemfinch (#1)')
self.assertRegexp('stats praise', r'currently 1')
self.assertNotError('add praise gives $who a cookie')
self.assertHelp('praise 1')
self.assertAction('praise 1 jemfinch', 'pets jemfinch (#1)')
self.assertAction('praise 2 jemfinch for being him',
'gives jemfinch a cookie for being him (#2)')
self.assertNotError('remove praise 1')
self.assertRegexp('stats praise', r'currently 1')
self.assertAction('praise jemfinch', 'gives jemfinch a cookie (#2)')
self.assertNotError('remove praise 2')
self.assertRegexp('stats praise', r'currently 0')
self.assertError('praise jemfinch')
def testPraise(self):
self.assertNotError('add praise pets $who')
self.assertHelp('praise')
self.assertResponse('praise jemfinch for being him',
'\x01ACTION pets jemfinch for being him '
'(#1)\x01')
self.assertResponse('praise jemfinch',
'\x01ACTION pets jemfinch (#1)\x01')
self.assertRegexp('stats praise', r'currently 1')
self.assertNotError('add praise gives $who a cookie')
self.assertHelp('praise 1')
self.assertResponse('praise 1 jemfinch',
'\x01ACTION pets jemfinch (#1)\x01')
self.assertResponse('praise 2 jemfinch for being him',
'\x01ACTION gives jemfinch a cookie for being '
'him (#2)\x01')
self.assertNotError('remove praise 1')
self.assertRegexp('stats praise', r'currently 1')
self.assertResponse('praise jemfinch',
'\x01ACTION gives jemfinch a cookie (#2)\x01')
self.assertNotError('remove praise 2')
self.assertRegexp('stats praise', r'currently 0')
self.assertError('praise jemfinch')
def testInfo(self):
self.assertNotError('add praise $who')
self.assertRegexp('info praise 1', r'Created by')
self.assertNotError('remove praise 1')
self.assertError('info fake 1')
def testInfo(self):
self.assertNotError('add praise $who')
self.assertRegexp('info praise 1', r'Created by')
self.assertNotError('remove praise 1')
self.assertError('info fake 1')
def testGet(self):
self.assertError('fundb get fake 1')
self.assertError('fundb get lart foo')
self.assertNotError('add praise pets $who')
self.assertResponse('fundb get praise 1', 'pets $who')
self.assertNotError('remove praise 1')
self.assertError('fundb get praise 1')
def testGet(self):
self.assertError('fundb get fake 1')
self.assertError('fundb get lart foo')
self.assertNotError('add praise pets $who')
self.assertResponse('fundb get praise 1', 'pets $who')
self.assertNotError('remove praise 1')
self.assertError('fundb get praise 1')
def testStats(self):
self.assertError('stats fake')
self.assertError('stats 1')
self.assertRegexp('stats praise', r'currently 0')
self.assertRegexp('stats lart', r'currently 0')
self.assertRegexp('stats insult', r'currently 0')
def testStats(self):
self.assertError('stats fake')
self.assertError('stats 1')
self.assertRegexp('stats praise', r'currently 0')
self.assertRegexp('stats lart', r'currently 0')
self.assertRegexp('stats insult', r'currently 0')
def testChange(self):
self.assertNotError('add praise teaches $who perl')
self.assertNotError('change praise 1 s/perl/python/')
self.assertAction('praise jemfinch', 'teaches jemfinch python (#1)')
self.assertNotError('remove praise 1')
def testChange(self):
self.assertNotError('add praise teaches $who perl')
self.assertNotError('change praise 1 s/perl/python/')
self.assertResponse('praise jemfinch',
'\x01ACTION teaches jemfinch python (#1)\x01')
self.assertNotError('remove praise 1')
def testConfig(self):
self.assertNotError('add praise teaches $who perl')
self.assertRegexp('praise jemfinch', r'\(#1\)')
conf.supybot.plugins.FunDB.showIds.setValue(False)
self.assertNotRegexp('praise jemfinch', r'\(#1\)')
def testConfig(self):
self.assertNotError('add praise teaches $who perl')
self.assertRegexp('praise jemfinch', r'\(#1\)')
conf.supybot.plugins.FunDB.showIds.setValue(False)
self.assertNotRegexp('praise jemfinch', r'\(#1\)')
def testLartPraiseReasonPeriod(self):
self.assertNotError('add lart kills $who')
self.assertNotRegexp('lart foo for bar.', r'\.')
self.assertNotError('add praise loves $who')
self.assertNotRegexp('praise for for bar.', r'\.')
def testLartPraiseReasonPeriod(self):
self.assertNotError('add lart kills $who')
self.assertNotRegexp('lart foo for bar.', r'\.')
self.assertNotError('add praise loves $who')
self.assertNotRegexp('praise for for bar.', r'\.')
# vim:set shiftwidth=4 tabstop=8 expandtab textwidth=78: