PgManager: Update RowDict, add write support.
authorRadek Brich <radek.brich@devl.cz>
Wed, 03 Apr 2013 15:57:36 +0200
changeset 81 f709b1c7ca0c
parent 80 b830c6f7b30a
child 82 7b82dc1fb6f5
PgManager: Update RowDict, add write support.
pgtoolkit/pgmanager.py
tests/test_rowdict.py
--- 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()
+