[pypy-svn] r69410 - in pypy/trunk/pypy/module/oracle: . test

[afa at codespeak.net]

Author: afa
Date: Thu Nov 19 10:12:21 2009
New Revision: 69410

Modified:
   pypy/trunk/pypy/module/oracle/interp_cursor.py
   pypy/trunk/pypy/module/oracle/test/test_select.py
Log:
implement cursor.fetchmany()


Modified: pypy/trunk/pypy/module/oracle/interp_cursor.py
==============================================================================
--- pypy/trunk/pypy/module/oracle/interp_cursor.py	(original)
+++ pypy/trunk/pypy/module/oracle/interp_cursor.py	Thu Nov 19 10:12:21 2009
@@ -733,6 +733,18 @@
         return space.w_None
     fetchone.unwrap_spec = ['self', ObjSpace]
 
+    def fetchmany(self, space, w_numRows=NoneNotWrapped):
+        if w_numRows is not None:
+            numRows = space.int_w(w_numRows)
+        else:
+            numRows = self.arraySize
+
+        # verify fetch can be performed
+        self._verifyFetch(space)
+
+        return self._multiFetch(space, limit=numRows)
+    fetchmany.unwrap_spec = ['self', ObjSpace, W_Root]
+
     def fetchall(self, space):
         # verify fetch can be performed
         self._verifyFetch(space)
@@ -811,6 +823,7 @@
 
         # fetch as many rows as possible
         while limit is None or rowNum < limit:
+            rowNum += 1
             if not self._moreRows(space):
                 break
             w_row = self._createRow(space)
@@ -1022,6 +1035,8 @@
                          unwrap_spec=W_Cursor.prepare.unwrap_spec),
     fetchone = interp2app(W_Cursor.fetchone,
                          unwrap_spec=W_Cursor.fetchone.unwrap_spec),
+    fetchmany = interp2app(W_Cursor.fetchmany,
+                         unwrap_spec=W_Cursor.fetchmany.unwrap_spec),
     fetchall = interp2app(W_Cursor.fetchall,
                          unwrap_spec=W_Cursor.fetchall.unwrap_spec),
     close = interp2app(W_Cursor.close,

Modified: pypy/trunk/pypy/module/oracle/test/test_select.py
==============================================================================
--- pypy/trunk/pypy/module/oracle/test/test_select.py	(original)
+++ pypy/trunk/pypy/module/oracle/test/test_select.py	Thu Nov 19 10:12:21 2009
@@ -28,9 +28,17 @@
         assert rows == zip(range(42))
         assert cur.rowcount == 42
 
+    def test_fetchmany(self):
+        cur = self.cnx.cursor()
+        cur.execute("select level-1 from dual connect by level-1<442")
+        rows = cur.fetchmany()
+        assert rows == zip(range(cur.arraysize))
+        rows = cur.fetchmany(3)
+        assert rows == zip(range(cur.arraysize, cur.arraysize+3))
+        assert cur.rowcount == cur.arraysize+3
+
     def test_iterator(self):
         cur = self.cnx.cursor()
-        # An Oracle trick to retrieve 42 lines
         cur.execute("select level-1 from dual connect by level-1<42")
         for i, row in enumerate(cur):
             assert row == (i,)