"""`True` if the cursor is closed."""
return self._closed
- @property
- def status(self) -> Optional[pq.ExecStatus]:
- # TODO: do we want this?
- res = self.pgresult
- return pq.ExecStatus(res.status) if res else None
-
@property
def query(self) -> Optional[bytes]:
"""The last query sent to the server, if available."""
assert w() is None
-def test_status(conn):
- cur = conn.cursor()
- assert cur.status is None
- cur.execute("reset all")
- assert cur.status == cur.ExecStatus.COMMAND_OK
- cur.execute("select 1")
- assert cur.status == cur.ExecStatus.TUPLES_OK
- cur.close()
- assert cur.status == cur.ExecStatus.TUPLES_OK
-
-
def test_execute_many_results(conn):
cur = conn.cursor()
assert cur.nextset() is None
def test_execute_empty_query(conn, query):
cur = conn.cursor()
cur.execute(query)
- assert cur.status == cur.ExecStatus.EMPTY_QUERY
+ assert cur.pgresult.status == cur.ExecStatus.EMPTY_QUERY
with pytest.raises(psycopg.ProgrammingError):
cur.fetchone()
assert w() is None
-async def test_status(aconn):
- cur = aconn.cursor()
- assert cur.status is None
- await cur.execute("reset all")
- assert cur.status == cur.ExecStatus.COMMAND_OK
- await cur.execute("select 1")
- assert cur.status == cur.ExecStatus.TUPLES_OK
- await cur.close()
- assert cur.status == cur.ExecStatus.TUPLES_OK
-
-
async def test_execute_many_results(aconn):
cur = aconn.cursor()
assert cur.nextset() is None
async def test_execute_empty_query(aconn, query):
cur = aconn.cursor()
await cur.execute(query)
- assert cur.status == cur.ExecStatus.EMPTY_QUERY
+ assert cur.pgresult.status == cur.ExecStatus.EMPTY_QUERY
with pytest.raises(psycopg.ProgrammingError):
await cur.fetchone()