loader. See :ref:`binary-data` for details.
:param row_factory: If specified override the `row_factory` set on the
connection. See :ref:`row-factories` for details.
+ :param scrollable: Specify the `~ServerCursor.scrollable` property of
+ the server-side cursor created.
+ :param withhold: Specify the `~ServerCursor.withhold` property of
+ the server-side cursor created.
:return: A cursor of the class specified by `cursor_factory` (or
`server_cursor_factory` if *name* is specified).
...` pattern is especially useful so that the cursor is closed at
the end of the block.
- .. automethod:: execute(query, params=None, *, scrollable=None, hold=False) -> ServerCursor
+ .. automethod:: execute(query, params=None, *, scrollable=None, withhold=False) -> ServerCursor
:param query: The query to execute.
:type query: `!str`, `!bytes`, or `sql.Composable`
:param scrollable: if `!True` make the cursor scrollable, if `!False`
not. if `!None` leave the choice to the server.
:type scrollable: `!Optional[bool]`
- :param hold: if `!True` allow the cursor to be used after the
- transaction creating it has committed.
- :type hold: `!bool`
+ :param withhold: if `!True` allow the cursor to be used after the
+ transaction creating it has committed.
+ :type withhold: `!bool`
Create a server cursor with given `name` and the *query* in argument.
If using :sql:`DECLARE` is not appropriate you can avoid to use
.. note:: You can close the cursor automatically using :samp:`async
with conn.cursor({name}): ...`
- .. automethod:: execute(query, params=None, *, scrollable=None, hold=False) -> AsyncServerCursor
+ .. automethod:: execute(query, params=None, *, scrollable=None, withhold=False) -> AsyncServerCursor
.. automethod:: executemany(query: Query, params_seq: Sequence[Args])
.. automethod:: fetchone
.. automethod:: fetchmany
cur: BaseCursor[ConnectionType, Row],
query: Query,
scrollable: Optional[bool],
- hold: bool,
+ withhold: bool,
) -> sql.Composable:
if isinstance(query, bytes):
if scrollable is not None:
parts.append(sql.SQL("scroll" if scrollable else "no scroll"))
parts.append(sql.SQL("cursor"))
- if hold:
+ if withhold:
parts.append(sql.SQL("with hold"))
parts.append(sql.SQL("for"))
parts.append(query)
params: Optional[Params] = None,
*,
scrollable: Optional[bool] = None,
- hold: bool = False,
+ withhold: bool = False,
) -> "ServerCursor[Row]":
"""
Open a cursor to execute a query to the database.
"""
query = self._helper._make_declare_statement(
- self, query, scrollable=scrollable, hold=hold
+ self, query, scrollable=scrollable, withhold=withhold
)
with self._conn.lock:
self._conn.wait(self._helper._declare_gen(self, query, params))
params: Optional[Params] = None,
*,
scrollable: Optional[bool] = None,
- hold: bool = False,
+ withhold: bool = False,
) -> "AsyncServerCursor[Row]":
query = self._helper._make_declare_statement(
- self, query, scrollable=scrollable, hold=hold
+ self, query, scrollable=scrollable, withhold=withhold
)
async with self._conn.lock:
await self._conn.wait(
curs.scroll(-1)
-@pytest.mark.parametrize("kwargs", [{}, {"hold": False}])
+@pytest.mark.parametrize("kwargs", [{}, {"withhold": False}])
def test_no_hold(conn, kwargs):
with pytest.raises(e.InvalidCursorName):
with conn.cursor("foo") as curs:
def test_hold(conn):
with conn.cursor("foo") as curs:
- curs.execute("select generate_series(0, 5)", hold=True)
+ curs.execute("select generate_series(0, 5)", withhold=True)
assert curs.fetchone() == (0,)
conn.commit()
assert curs.fetchone() == (1,)
await curs.scroll(-1)
-@pytest.mark.parametrize("kwargs", [{}, {"hold": False}])
+@pytest.mark.parametrize("kwargs", [{}, {"withhold": False}])
async def test_no_hold(aconn, kwargs):
with pytest.raises(e.InvalidCursorName):
async with aconn.cursor("foo") as curs:
async def test_hold(aconn):
async with aconn.cursor("foo") as curs:
- await curs.execute("select generate_series(0, 5)", hold=True)
+ await curs.execute("select generate_series(0, 5)", withhold=True)
assert await curs.fetchone() == (0,)
await aconn.commit()
assert await curs.fetchone() == (1,)