`!None` if the current resultset didn't return tuples.
"""
res = self.pgresult
- if not (res and res.nfields):
+
+ # We return columns if we have nfields, but also if we don't but
+ # the query said we got tuples (mostly to handle the super useful
+ # query "SELECT ;"
+ if res and (
+ res.nfields
+ or res.status == ExecStatus.TUPLES_OK
+ or res.status == ExecStatus.SINGLE_TUPLE
+ ):
+ return [Column(self, i) for i in range(res.nfields)]
+ else:
return None
- return [Column(self, i) for i in range(res.nfields)]
@property
def rowcount(self) -> int:
def test_row_factory(conn):
cur = conn.cursor(row_factory=my_row_factory)
+
+ cur.execute("reset search_path")
+ with pytest.raises(psycopg.ProgrammingError):
+ cur.fetchone()
+
cur.execute("select 'foo' as bar")
(r,) = cur.fetchone()
assert r == "FOObar"
assert next(it).a == 2
+def test_stream_no_col(conn):
+ cur = conn.cursor()
+ it = iter(cur.stream("select"))
+ assert list(it) == [()]
+
+
@pytest.mark.parametrize(
"query",
[
unpickled = pickle.loads(pickled)
assert [tuple(d) for d in description] == [tuple(d) for d in unpickled]
+ def test_no_col_query(self, conn):
+ cur = conn.execute("select")
+ assert cur.description == []
+ assert cur.fetchall() == [()]
+
def test_str(conn):
cur = conn.cursor()
assert not cur.nextset()
+def test_no_result(conn):
+ with conn.cursor("foo") as cur:
+ cur.execute("select generate_series(1, %s) as bar where false", (3,))
+ assert len(cur.description) == 1
+ assert cur.fetchall() == []
+
+
def test_row_factory(conn):
n = 0
assert not cur.nextset()
+async def test_no_result(aconn):
+ async with aconn.cursor("foo") as cur:
+ await cur.execute(
+ "select generate_series(1, %s) as bar where false", (3,)
+ )
+ assert len(cur.description) == 1
+ assert (await cur.fetchall()) == []
+
+
async def test_row_factory(aconn):
n = 0