mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-11-26 13:22:51 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

Make gdbm and dumbdbm use byte strings. Updated their tests.

Browse source
This commit is contained in:
Guido van Rossum 2007-05-23 20:51:02 +00:00
parent 517bcfeb6b
commit 6252e10ed9
7 changed files with 68 additions and 61 deletions
Download patch file Download diff file Expand all files Collapse all files

29
Lib/dumbdbm.py
View file

@ -21,12 +21,11 @@ is read when the database is opened, and some updates rewrite the whole index)
""" """
import io as _io
import os as _os import os as _os
import __builtin__ import __builtin__
import UserDict import UserDict
_open = __builtin__.open
_BLOCKSIZE = 512 _BLOCKSIZE = 512
error = IOError # For anydbm error = IOError # For anydbm
@ -42,7 +41,7 @@ class _Database(UserDict.DictMixin):
# _commit() finish successfully, we can't ignore shutdown races # _commit() finish successfully, we can't ignore shutdown races
# here, and _commit() must not reference any globals. # here, and _commit() must not reference any globals.
_os = _os # for _commit() _os = _os # for _commit()
_open = _open # for _commit() _io = _io # for _commit()
def __init__(self, filebasename, mode): def __init__(self, filebasename, mode):
self._mode = mode self._mode = mode
@ -66,9 +65,9 @@ class _Database(UserDict.DictMixin):
# Mod by Jack: create data file if needed # Mod by Jack: create data file if needed
try: try:
f = _open(self._datfile, 'r') f = _io.open(self._datfile, 'r')
except IOError: except IOError:
f = _open(self._datfile, 'w') f = _io.open(self._datfile, 'w')
self._chmod(self._datfile) self._chmod(self._datfile)
f.close() f.close()
self._update() self._update()
@ -77,7 +76,7 @@ class _Database(UserDict.DictMixin):
def _update(self): def _update(self):
self._index = {} self._index = {}
try: try:
f = _open(self._dirfile) f = _io.open(self._dirfile, 'r')
except IOError: except IOError:
pass pass
else: else:
@ -107,7 +106,7 @@ class _Database(UserDict.DictMixin):
except self._os.error: except self._os.error:
pass pass
f = self._open(self._dirfile, 'w') f = self._io.open(self._dirfile, 'w')
self._chmod(self._dirfile) self._chmod(self._dirfile)
for key, pos_and_siz_pair in self._index.items(): for key, pos_and_siz_pair in self._index.items():
f.write("%r, %r\n" % (key, pos_and_siz_pair)) f.write("%r, %r\n" % (key, pos_and_siz_pair))
@ -117,7 +116,7 @@ class _Database(UserDict.DictMixin):
def __getitem__(self, key): def __getitem__(self, key):
pos, siz = self._index[key] # may raise KeyError pos, siz = self._index[key] # may raise KeyError
f = _open(self._datfile, 'rb') f = _io.open(self._datfile, 'rb')
f.seek(pos) f.seek(pos)
dat = f.read(siz) dat = f.read(siz)
f.close() f.close()
@ -128,11 +127,11 @@ class _Database(UserDict.DictMixin):
# to get to an aligned offset. Return pair # to get to an aligned offset. Return pair
# (starting offset of val, len(val)) # (starting offset of val, len(val))
def _addval(self, val): def _addval(self, val):
f = _open(self._datfile, 'rb+') f = _io.open(self._datfile, 'rb+')
f.seek(0, 2) f.seek(0, 2)
pos = int(f.tell()) pos = int(f.tell())
npos = ((pos + _BLOCKSIZE - 1) // _BLOCKSIZE) * _BLOCKSIZE npos = ((pos + _BLOCKSIZE - 1) // _BLOCKSIZE) * _BLOCKSIZE
f.write('\0'*(npos-pos)) f.write(b'\0'*(npos-pos))
pos = npos pos = npos
f.write(val) f.write(val)
f.close() f.close()
@ -143,7 +142,7 @@ class _Database(UserDict.DictMixin):
# pos to hold val, without overwriting some other value. Return # pos to hold val, without overwriting some other value. Return
# pair (pos, len(val)). # pair (pos, len(val)).
def _setval(self, pos, val): def _setval(self, pos, val):
f = _open(self._datfile, 'rb+') f = _io.open(self._datfile, 'rb+')
f.seek(pos) f.seek(pos)
f.write(val) f.write(val)
f.close() f.close()
@ -154,14 +153,16 @@ class _Database(UserDict.DictMixin):
# the in-memory index dict, and append one to the directory file. # the in-memory index dict, and append one to the directory file.
def _addkey(self, key, pos_and_siz_pair): def _addkey(self, key, pos_and_siz_pair):
self._index[key] = pos_and_siz_pair self._index[key] = pos_and_siz_pair
f = _open(self._dirfile, 'a') f = _io.open(self._dirfile, 'a')
self._chmod(self._dirfile) self._chmod(self._dirfile)
f.write("%r, %r\n" % (key, pos_and_siz_pair)) f.write("%r, %r\n" % (key, pos_and_siz_pair))
f.close() f.close()
def __setitem__(self, key, val): def __setitem__(self, key, val):
if not type(key) == type('') == type(val): if not isinstance(key, basestring):
raise TypeError, "keys and values must be strings" raise TypeError("keys must be strings")
if not isinstance(val, (str8, bytes)):
raise TypeError("values must be byte strings")
if key not in self._index: if key not in self._index:
self._addkey(key, self._addval(val)) self._addkey(key, self._addval(val))
else: else:

16
Lib/test/test_anydbm.py
View file

@ -21,13 +21,13 @@ def _delete_files():
pass pass
class AnyDBMTestCase(unittest.TestCase): class AnyDBMTestCase(unittest.TestCase):
_dict = {'0': '', _dict = {'0': b'',
'a': 'Python:', 'a': b'Python:',
'b': 'Programming', 'b': b'Programming',
'c': 'the', 'c': b'the',
'd': 'way', 'd': b'way',
'f': 'Guido', 'f': b'Guido',
'g': 'intended' 'g': b'intended',
} }
def __init__(self, *args): def __init__(self, *args):
@ -44,7 +44,7 @@ class AnyDBMTestCase(unittest.TestCase):
def test_anydbm_modification(self): def test_anydbm_modification(self):
self.init_db() self.init_db()
f = anydbm.open(_fname, 'c') f = anydbm.open(_fname, 'c')
self._dict['g'] = f['g'] = "indented" self._dict['g'] = f['g'] = b"indented"
self.read_helper(f) self.read_helper(f)
f.close() f.close()

43
Lib/test/test_dumbdbm.py
View file

@ -3,6 +3,7 @@
Original by Roger E. Masse Original by Roger E. Masse
""" """
import io
import os import os
import unittest import unittest
import dumbdbm import dumbdbm
@ -18,13 +19,13 @@ def _delete_files():
pass pass
class DumbDBMTestCase(unittest.TestCase): class DumbDBMTestCase(unittest.TestCase):
_dict = {'0': '', _dict = {'0': b'',
'a': 'Python:', 'a': b'Python:',
'b': 'Programming', 'b': b'Programming',
'c': 'the', 'c': b'the',
'd': 'way', 'd': b'way',
'f': 'Guido', 'f': b'Guido',
'g': 'intended' 'g': b'intended',
} }
def __init__(self, *args): def __init__(self, *args):
@ -64,15 +65,15 @@ class DumbDBMTestCase(unittest.TestCase):
def test_close_twice(self): def test_close_twice(self):
f = dumbdbm.open(_fname) f = dumbdbm.open(_fname)
f['a'] = 'b' f['a'] = b'b'
self.assertEqual(f['a'], 'b') self.assertEqual(f['a'], b'b')
f.close() f.close()
f.close() f.close()
def test_dumbdbm_modification(self): def test_dumbdbm_modification(self):
self.init_db() self.init_db()
f = dumbdbm.open(_fname, 'w') f = dumbdbm.open(_fname, 'w')
self._dict['g'] = f['g'] = "indented" self._dict['g'] = f['g'] = b"indented"
self.read_helper(f) self.read_helper(f)
f.close() f.close()
@ -91,29 +92,29 @@ class DumbDBMTestCase(unittest.TestCase):
def test_write_write_read(self): def test_write_write_read(self):
# test for bug #482460 # test for bug #482460
f = dumbdbm.open(_fname) f = dumbdbm.open(_fname)
f['1'] = 'hello' f['1'] = b'hello'
f['1'] = 'hello2' f['1'] = b'hello2'
f.close() f.close()
f = dumbdbm.open(_fname) f = dumbdbm.open(_fname)
self.assertEqual(f['1'], 'hello2') self.assertEqual(f['1'], b'hello2')
f.close() f.close()
def test_line_endings(self): def test_line_endings(self):
# test for bug #1172763: dumbdbm would die if the line endings # test for bug #1172763: dumbdbm would die if the line endings
# weren't what was expected. # weren't what was expected.
f = dumbdbm.open(_fname) f = dumbdbm.open(_fname)
f['1'] = 'hello' f['1'] = b'hello'
f['2'] = 'hello2' f['2'] = b'hello2'
f.close() f.close()
# Mangle the file by adding \r before each newline # Mangle the file by adding \r before each newline
data = open(_fname + '.dir').read() data = io.open(_fname + '.dir', 'rb').read()
data = data.replace('\n', '\r\n') data = data.replace(b'\n', b'\r\n')
open(_fname + '.dir', 'wb').write(data) io.open(_fname + '.dir', 'wb').write(data)
f = dumbdbm.open(_fname) f = dumbdbm.open(_fname)
self.assertEqual(f['1'], 'hello') self.assertEqual(f['1'], b'hello')
self.assertEqual(f['2'], 'hello2') self.assertEqual(f['2'], b'hello2')
def read_helper(self, f): def read_helper(self, f):
@ -147,7 +148,7 @@ class DumbDBMTestCase(unittest.TestCase):
del d[k] del d[k]
del f[k] del f[k]
else: else:
v = random.choice('abc') * random.randrange(10000) v = random.choice((b'a', b'b', b'c')) * random.randrange(10000)
d[k] = v d[k] = v
f[k] = v f[k] = v
self.assertEqual(f[k], v) self.assertEqual(f[k], v)

2
Lib/test/test_gdbm.py
View file

@ -12,7 +12,7 @@ filename = TESTFN
g = gdbm.open(filename, 'c') g = gdbm.open(filename, 'c')
verify(g.keys() == []) verify(g.keys() == [])
g['a'] = 'b' g['a'] = 'b'
g['12345678910'] = '019237410982340912840198242' g['12345678910'] = b'019237410982340912840198242'
a = g.keys() a = g.keys()
if verbose: if verbose:
print('Test gdbm file keys: ', a) print('Test gdbm file keys: ', a)

2
Lib/test/test_whichdb.py
View file

@ -51,7 +51,7 @@ for name in anydbm._names:
self.assertEqual(name, whichdb.whichdb(_fname)) self.assertEqual(name, whichdb.whichdb(_fname))
# Now add a key # Now add a key
f = mod.open(_fname, 'w') f = mod.open(_fname, 'w')
f["1"] = "1" f["1"] = b"1"
f.close() f.close()
self.assertEqual(name, whichdb.whichdb(_fname)) self.assertEqual(name, whichdb.whichdb(_fname))
setattr(WhichDBTestCase,"test_whichdb_%s" % name, test_whichdb_name) setattr(WhichDBTestCase,"test_whichdb_%s" % name, test_whichdb_name)

13
Lib/whichdb.py
View file

@ -1,6 +1,7 @@
# !/usr/bin/env python # !/usr/bin/env python
"""Guess which db package to use to open a db file.""" """Guess which db package to use to open a db file."""
import io
import os import os
import struct import struct
import sys import sys
@ -29,18 +30,18 @@ def whichdb(filename):
# Check for dbm first -- this has a .pag and a .dir file # Check for dbm first -- this has a .pag and a .dir file
try: try:
f = open(filename + os.extsep + "pag", "rb") f = io.open(filename + os.extsep + "pag", "rb")
f.close() f.close()
# dbm linked with gdbm on OS/2 doesn't have .dir file # dbm linked with gdbm on OS/2 doesn't have .dir file
if not (dbm.library == "GNU gdbm" and sys.platform == "os2emx"): if not (dbm.library == "GNU gdbm" and sys.platform == "os2emx"):
f = open(filename + os.extsep + "dir", "rb") f = io.open(filename + os.extsep + "dir", "rb")
f.close() f.close()
return "dbm" return "dbm"
except IOError: except IOError:
# some dbm emulations based on Berkeley DB generate a .db file # some dbm emulations based on Berkeley DB generate a .db file
# some do not, but they should be caught by the dbhash checks # some do not, but they should be caught by the dbhash checks
try: try:
f = open(filename + os.extsep + "db", "rb") f = io.open(filename + os.extsep + "db", "rb")
f.close() f.close()
# guarantee we can actually open the file using dbm # guarantee we can actually open the file using dbm
# kind of overkill, but since we are dealing with emulations # kind of overkill, but since we are dealing with emulations
@ -60,9 +61,9 @@ def whichdb(filename):
# dumbdbm files with no keys are empty # dumbdbm files with no keys are empty
if size == 0: if size == 0:
return "dumbdbm" return "dumbdbm"
f = open(filename + os.extsep + "dir", "rb") f = io.open(filename + os.extsep + "dir", "rb")
try: try:
if f.read(1) in ("'", '"'): if f.read(1) in (b"'", b'"'):
return "dumbdbm" return "dumbdbm"
finally: finally:
f.close() f.close()
@ -71,7 +72,7 @@ def whichdb(filename):
# See if the file exists, return None if not # See if the file exists, return None if not
try: try:
f = open(filename, "rb") f = io.open(filename, "rb")
except IOError: except IOError:
return None return None

16
Modules/gdbmmodule.c
View file

@ -127,11 +127,10 @@ dbm_subscript(dbmobject *dp, register PyObject *key)
} }
drec = gdbm_fetch(dp->di_dbm, krec); drec = gdbm_fetch(dp->di_dbm, krec);
if (drec.dptr == 0) { if (drec.dptr == 0) {
PyErr_SetString(PyExc_KeyError, PyErr_SetObject(PyExc_KeyError, key);
PyString_AS_STRING((PyStringObject *)key));
return NULL; return NULL;
} }
v = PyString_FromStringAndSize(drec.dptr, drec.dsize); v = PyBytes_FromStringAndSize(drec.dptr, drec.dsize);
free(drec.dptr); free(drec.dptr);
return v; return v;
} }
@ -154,15 +153,14 @@ dbm_ass_sub(dbmobject *dp, PyObject *v, PyObject *w)
dp->di_size = -1; dp->di_size = -1;
if (w == NULL) { if (w == NULL) {
if (gdbm_delete(dp->di_dbm, krec) < 0) { if (gdbm_delete(dp->di_dbm, krec) < 0) {
PyErr_SetString(PyExc_KeyError, PyErr_SetObject(PyExc_KeyError, v);
PyString_AS_STRING((PyStringObject *)v));
return -1; return -1;
} }
} }
else { else {
if (!PyArg_Parse(w, "s#", &drec.dptr, &drec.dsize)) { if (!PyArg_Parse(w, "s#", &drec.dptr, &drec.dsize)) {
PyErr_SetString(PyExc_TypeError, PyErr_SetString(PyExc_TypeError,
"gdbm mappings have string elements only"); "gdbm mappings have byte string elements only");
return -1; return -1;
} }
errno = 0; errno = 0;
@ -198,6 +196,7 @@ dbm_close(register dbmobject *dp, PyObject *unused)
return Py_None; return Py_None;
} }
/* XXX Should return a set or a set view */
PyDoc_STRVAR(dbm_keys__doc__, PyDoc_STRVAR(dbm_keys__doc__,
"keys() -> list_of_keys\n\ "keys() -> list_of_keys\n\
Get a list of all keys in the database."); Get a list of all keys in the database.");
@ -252,6 +251,11 @@ dbm_contains(PyObject *self, PyObject *arg)
"GDBM object has already been closed"); "GDBM object has already been closed");
return -1; return -1;
} }
if (PyUnicode_Check(arg)) {
arg = _PyUnicode_AsDefaultEncodedString(arg, NULL);
if (arg == NULL)
return -1;
}
if (!PyString_Check(arg)) { if (!PyString_Check(arg)) {
PyErr_Format(PyExc_TypeError, PyErr_Format(PyExc_TypeError,
"gdbm key must be string, not %.100s", "gdbm key must be string, not %.100s",