Test and document the named cursor stealing technique

This commit is contained in:
Daniele Varrazzo 2011-10-14 23:17:24 +01:00
parent dde4c0de3d
commit 37a9eb3615
2 changed files with 47 additions and 1 deletions

View File

@ -562,7 +562,7 @@ subsequently handled using :sql:`MOVE`, :sql:`FETCH` and :sql:`CLOSE` commands.
Psycopg wraps the database server side cursor in *named cursors*. A named
cursor is created using the `~connection.cursor()` method specifying the
`name` parameter. Such cursor will behave mostly like a regular cursor,
*name* parameter. Such cursor will behave mostly like a regular cursor,
allowing the user to move in the dataset using the `~cursor.scroll()`
method and to read the data using `~cursor.fetchone()` and
`~cursor.fetchmany()` methods.
@ -588,6 +588,38 @@ will be eventually be closed. Also note that while :sql:`WITH HOLD` cursors
lifetime extends well after `~connection.commit()`, calling
`~connection.rollback()` will automatically close the cursor.
.. note::
It is also possible to use a named cursor to consume a cursor created
in some other way than using the |DECLARE| executed by
`~cursor.execute()`. For example, you may have a PL/pgSQL function
returning a cursor::
CREATE FUNCTION reffunc(refcursor) RETURNS refcursor AS $$
BEGIN
OPEN $1 FOR SELECT col FROM test;
RETURN $1;
END;
$$ LANGUAGE plpgsql;
You can read the cursor content by calling the function with a regular,
non-named, Psycopg cursor:
.. code-block:: python
cur1 = conn.cursor()
cur1.callproc('reffunc', ['curname'])
and then use a named cursor in the same transaction to "steal the cursor":
.. code-block:: python
cur2 = conn.cursor('curname')
for record in cur2: # or cur2.fetchone, fetchmany...
# do something with record
pass
.. |DECLARE| replace:: :sql:`DECLARE`
.. _DECLARE: http://www.postgresql.org/docs/9.0/static/sql-declare.html

View File

@ -258,6 +258,20 @@ class CursorTests(unittest.TestCase):
self.assertEqual(c.precision, None)
self.assertEqual(c.scale, None)
@skip_before_postgres(8, 0)
def test_named_cursor_stealing(self):
# you can use a named cursor to iterate on a refcursor created
# somewhere else
cur1 = self.conn.cursor()
cur1.execute("DECLARE test CURSOR WITHOUT HOLD "
" FOR SELECT generate_series(1,7)")
cur2 = self.conn.cursor('test')
# can call fetch without execute
self.assertEqual((1,), cur2.fetchone())
self.assertEqual([(2,), (3,), (4,)], cur2.fetchmany(3))
self.assertEqual([(5,), (6,), (7,)], cur2.fetchall())
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)