# HG changeset patch # User Radek Brich # Date 1364997456 -7200 # Node ID f709b1c7ca0c4202f797729a50feb5a099809d67 # Parent b830c6f7b30aa522dd6e3d68f941f36171304aec PgManager: Update RowDict, add write support. diff -r b830c6f7b30a -r f709b1c7ca0c pgtoolkit/pgmanager.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): diff -r b830c6f7b30a -r f709b1c7ca0c tests/test_rowdict.py --- 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() +