PgManager: Update RowDict, add write support.
--- a/pgtoolkit/pgmanager.py Thu Mar 21 17:40:11 2013 +0100
+++ b/pgtoolkit/pgmanager.py Wed Apr 03 15:57:36 2013 +0200
@@ -104,53 +104,39 @@
self.pool_size = pool_size
-class RowDict:
- """Special read-only dictionary used for rows returned from queries.
-
- Initialization is same as for dict:
- row = RowDict([('id', 123), ('name', 'hello')])
-
- Allows key and attribute access to contained items:
- row['id']
- row.id
+class RowDict(OrderedDict):
+ """Special dictionary used for rows returned from queries.
Items keep order in which columns where returned from database.
- Tuple style access is also supported:
- row[0]
- id, name = row
+ It supports three styles of access:
+
+ Dict style:
+ row['id']
+ for key in row:
+ ...
+
+ Object style (only works if column name does not collide with any method name):
+ row.id
+
+ Tuple style:
+ row[0]
+ id, name = row.values()
"""
- def __init__(self, data):
- self._dict = OrderedDict(data)
-
def __getitem__(self, key):
if isinstance(key, int):
- return tuple(self._dict.values())[key]
- return self._dict[key]
+ return tuple(self.values())[key]
+ else:
+ return OrderedDict.__getitem__(self, key)
def __getattr__(self, key):
try:
- return self._dict[key]
+ return self[key]
except KeyError:
raise AttributeError(key)
- def __contains__(self, key):
- return key in self._dict
-
- def keys(self):
- return self._dict.keys()
-
- def values(self):
- return self._dict.values()
-
- def items(self):
- return self._dict.items()
-
- def __repr__(self):
- return 'RowDict(%s)' % str(self._dict.items())
-
class Cursor(psycopg2.extensions.cursor):
@@ -491,7 +477,7 @@
return psycopg2.extensions.ISOLATION_LEVEL_READ_COMMITTED
if level.lower() == 'serializable':
return psycopg2.extensions.ISOLATION_LEVEL_SERIALIZABLE
- raise PgManagerError('Unknown isolation level name: "%s"', level)
+ raise PgManagerError('Unknown isolation level name: "%s"' % level)
return level
def _check_fork(self):
--- a/tests/test_rowdict.py Thu Mar 21 17:40:11 2013 +0100
+++ b/tests/test_rowdict.py Wed Apr 03 15:57:36 2013 +0200
@@ -23,9 +23,15 @@
self.assertEqual(self.rowdict[0], 123)
def test_rowdict_unpack(self):
- _id, name = self.rowdict
+ _id, name = self.rowdict.values()
self.assertEqual(name, 'hello')
+ def test_rowdict_repr(self):
+ repr_text = "RowDict([('id', 123), ('name', 'hello')])"
+ self.assertEqual(repr(self.rowdict), repr_text)
+ self.assertEqual(str(self.rowdict), repr_text)
+
if __name__ == '__main__':
unittest.main()
+