Such database version doesn't allow to emit unknown oid literals.
However specifying a different type for unknown, such as text, will fail
as soon as the value is used in a typed context, which is a far more
useful use case.
def test_execute_sequence(conn):
cur = conn.cursor()
- rv = cur.execute("select %s, %s, %s", [1, "foo", None])
+ cast = "::text" if conn.pgconn.server_version < 100000 else ""
+ rv = cur.execute(f"select %s, %s, %s{cast}", [1, "foo", None])
assert rv is cur
assert len(cur._results) == 1
assert cur.pgresult.get_value(0, 0) == b"1"
def test_fetchone(conn):
cur = conn.cursor()
- cur.execute("select %s, %s, %s", [1, "foo", None])
+ cast = "::text" if conn.pgconn.server_version < 100000 else ""
+ cur.execute(f"select %s, %s, %s{cast}", [1, "foo", None])
assert cur.pgresult.fformat(0) == 0
row = cur.fetchone()
def test_execute_binary_result(conn):
cur = conn.cursor(format=psycopg3.pq.Format.BINARY)
- cur.execute("select %s, %s", ["foo", None])
+ cast = "::text" if conn.pgconn.server_version < 100000 else ""
+ cur.execute(f"select %s, %s{cast}", ["foo", None])
assert cur.pgresult.fformat(0) == 1
row = cur.fetchone()
async def test_execute_sequence(aconn):
cur = aconn.cursor()
- rv = await cur.execute("select %s, %s, %s", [1, "foo", None])
+ cast = "::text" if aconn.pgconn.server_version < 100000 else ""
+ rv = await cur.execute(f"select %s, %s, %s{cast}", [1, "foo", None])
assert rv is cur
assert len(cur._results) == 1
assert cur.pgresult.get_value(0, 0) == b"1"
async def test_fetchone(aconn):
cur = aconn.cursor()
- await cur.execute("select %s, %s, %s", [1, "foo", None])
+ cast = "::text" if aconn.pgconn.server_version < 100000 else ""
+ await cur.execute(f"select %s, %s, %s{cast}", [1, "foo", None])
assert cur.pgresult.fformat(0) == 0
row = await cur.fetchone()
async def test_execute_binary_result(aconn):
cur = aconn.cursor(format=psycopg3.pq.Format.BINARY)
- await cur.execute("select %s, %s", ["foo", None])
+ cast = "::text" if aconn.pgconn.server_version < 100000 else ""
+ await cur.execute(f"select %s, %s{cast}", ["foo", None])
assert cur.pgresult.fformat(0) == 1
row = await cur.fetchone()
def test_roundtrip_bool(conn, b, fmt_in, fmt_out):
cur = conn.cursor(format=fmt_out)
ph = "%s" if fmt_in == Format.TEXT else "%b"
- result = cur.execute(f"select {ph}", (b,)).fetchone()[0]
+ cast = "" if conn.pgconn.server_version > 100000 else "::bool"
+ result = cur.execute(f"select {ph}{cast}", (b,)).fetchone()[0]
assert cur.pgresult.fformat(0) == fmt_out
if b is not None:
assert cur.pgresult.ftype(0) == builtins["bool"].oid