Build the namedtuple only once per execution, not once per fetch.

ChangeLog

@@ -1,3 +1,8 @@
+2010-11-11  Daniele Varrazzo  <daniele.varrazzo@gmail.com>
+
+	* lib/extras.py: build the namedtuple only once per execution, not once
+	per fetch.
+
 2010-11-10  Daniele Varrazzo  <daniele.varrazzo@gmail.com>
 
 	* psycopg/green.c: functions unused outside the module marked static.

lib/extras.py

@@ -259,19 +259,39 @@ class NamedTupleCursor(_cursor):
     .. |namedtuple| replace:: `!namedtuple`
     .. __: http://docs.python.org/release/2.6/library/collections.html#collections.namedtuple
     """
+    Record = None
+
+    def execute(self, query, vars=None):
+        self.Record = None
+        return _cursor.execute(self, query, vars)
+
+    def executemany(self, query, vars):
+        self.Record = None
+        return _cursor.executemany(self, vars)
+
+    def callproc(self, procname, vars=None):
+        self.Record = None
+        return _cursor.callproc(self, procname, vars)
+
     def fetchone(self):
         t = _cursor.fetchone(self)
         if t is not None:
-            nt = self._make_nt()
+            nt = self.Record
+            if nt is None:
+                nt = self.Record = self._make_nt()
             return nt(*t)
 
     def fetchmany(self, size=None):
-        nt = self._make_nt()
+        nt = self.Record
+        if nt is None:
+            nt = self.Record = self._make_nt()
         ts = _cursor.fetchmany(self, size)
         return [nt(*t) for t in ts]
 
     def fetchall(self):
-        nt = self._make_nt()
+        nt = self.Record
+        if nt is None:
+            nt = self.Record = self._make_nt()
         ts = _cursor.fetchall(self)
         return [nt(*t) for t in ts]
 

tests/extras_dictcursor.py

@@ -207,6 +207,51 @@ class NamedTupleCursorTest(unittest.TestCase):
             # skip the test
             pass
 
+    @if_has_namedtuple
+    def test_record_updated(self):
+        curs = self.conn.cursor()
+        curs.execute("select 1 as foo;")
+        r = curs.fetchone()
+        self.assertEqual(r.foo, 1)
+
+        curs.execute("select 2 as bar;")
+        r = curs.fetchone()
+        self.assertEqual(r.bar, 2)
+        self.assertRaises(AttributeError, getattr, r, 'foo')
+
+    @if_has_namedtuple
+    def test_minimal_generation(self):
+        # Instrument the class to verify it gets called the minimum number of times.
+        from psycopg2.extras import NamedTupleCursor
+        f_orig = NamedTupleCursor._make_nt
+        calls = [0]
+        def f_patched(self_):
+            calls[0] += 1
+            return f_orig(self_)
+
+        NamedTupleCursor._make_nt = f_patched
+
+        try:
+            curs = self.conn.cursor()
+            curs.execute("select * from nttest order by 1")
+            curs.fetchone()
+            curs.fetchone()
+            curs.fetchone()
+            self.assertEqual(1, calls[0])
+
+            curs.execute("select * from nttest order by 1")
+            curs.fetchone()
+            curs.fetchall()
+            self.assertEqual(2, calls[0])
+
+            curs.execute("select * from nttest order by 1")
+            curs.fetchone()
+            curs.fetchmany(1)
+            self.assertEqual(3, calls[0])
+
+        finally:
+            NamedTupleCursor._make_nt = f_orig
+
 
 def test_suite():
     return unittest.TestLoader().loadTestsFromName(__name__)