From: Daniele Varrazzo Date: Thu, 5 Nov 2020 14:40:17 +0000 (+0100) Subject: Use oid 0 for unknown typed and string casting X-Git-Tag: 3.0.dev0~397 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=d65a93d495cfef66e95d3f46180dc5ac5a046cdd;p=thirdparty%2Fpsycopg.git Use oid 0 for unknown typed and string casting Using unknown (0) oid instead of text (25) works better to throw normal strings at typed targets (e.g. columns) and expect them to work, a behaviour similar to client-side binding. However Postgres < 10 refuse to emit columns with unknown oid. --- diff --git a/psycopg3/psycopg3/adapt.py b/psycopg3/psycopg3/adapt.py index b911e74d7..f0b9e14fd 100644 --- a/psycopg3/psycopg3/adapt.py +++ b/psycopg3/psycopg3/adapt.py @@ -41,7 +41,7 @@ class Dumper: @property def oid(self) -> int: - return TEXT_OID + return 0 @classmethod def register( diff --git a/psycopg3/psycopg3/types/array.py b/psycopg3/psycopg3/types/array.py index 3b08031b3..955944eed 100644 --- a/psycopg3/psycopg3/types/array.py +++ b/psycopg3/psycopg3/types/array.py @@ -111,7 +111,7 @@ class ListBinaryDumper(BaseListDumper): data: List[bytes] = [b"", b""] # placeholders to avoid a resize dims: List[int] = [] hasnull = 0 - oid: Optional[int] = None + oid = 0 def calc_dims(L: List[Any]) -> None: if isinstance(L, self.src): @@ -134,7 +134,7 @@ class ListBinaryDumper(BaseListDumper): ad = dumper.dump(item) data.append(_struct_len.pack(len(ad))) data.append(ad) - if oid is None: + if not oid: oid = dumper.oid else: hasnull = 1 @@ -149,7 +149,7 @@ class ListBinaryDumper(BaseListDumper): dump_list(obj, 0) - if oid is None: + if not oid: oid = TEXT_OID self._array_oid = self._get_array_oid(oid) diff --git a/psycopg3/psycopg3/types/text.py b/psycopg3/psycopg3/types/text.py index c5988778c..6ad6ca45c 100644 --- a/psycopg3/psycopg3/types/text.py +++ b/psycopg3/psycopg3/types/text.py @@ -22,7 +22,7 @@ class _StringDumper(Dumper): super().__init__(src, context) self._encode: EncodeFunc - if self.connection is not None: + if self.connection: if self.connection.client_encoding != "SQL_ASCII": self._encode = self.connection.codec.encode else: diff --git a/tests/test_adapt.py b/tests/test_adapt.py index e50b22003..6b5b4b202 100644 --- a/tests/test_adapt.py +++ b/tests/test_adapt.py @@ -1,4 +1,6 @@ import pytest + +import psycopg3 from psycopg3.adapt import Transformer, Format, Dumper, Loader from psycopg3.oids import builtins @@ -17,7 +19,7 @@ def test_dump(data, format, result, type): t = Transformer() dumper = t.get_dumper(data, format) assert dumper.dump(data) == result - assert dumper.oid == builtins[type].oid + assert dumper.oid == 0 if type == "text" else builtins[type].oid @pytest.mark.parametrize( @@ -67,7 +69,9 @@ def test_dump_subclass(conn, fmt_out): pass cur = conn.cursor() - cur.execute("select %s, %b", [MyString("hello"), MyString("world")]) + cur.execute( + "select %s::text, %b::text", [MyString("hello"), MyString("world")] + ) assert cur.fetchone() == ("hello", "world") @@ -147,10 +151,36 @@ def test_none_type_argument(conn, fmt_in): assert cur.fetchone()[0] +@pytest.mark.parametrize("fmt_in", [Format.TEXT, Format.BINARY]) +def test_return_untyped(conn, fmt_in): + # Analyze and check for changes using strings in untyped/typed contexts + cur = conn.cursor() + # Currently string are passed as unknown oid to libpq. This is because + # unknown is more easily cast by postgres to different types (see jsonb + # later). However Postgres < 10 refuses to emit unknown types. + if conn.pgconn.server_version > 100000: + cur.execute("select %s, %s", ["hello", 10]) + assert cur.fetchone() == ("hello", 10) + else: + with pytest.raises(psycopg3.errors.IndeterminateDatatype): + cur.execute("select %s, %s", ["hello", 10]) + conn.rollback() + cur.execute("select %s::text, %s", ["hello", 10]) + assert cur.fetchone() == ("hello", 10) + + # It would be nice if above all postgres version behaved consistently. + # However this below shouldn't break either. + cur.execute("create table testjson(data jsonb)") + cur.execute("insert into testjson (data) values (%s)", ["{}"]) + assert cur.execute("select data from testjson").fetchone() == ({},) + + def make_dumper(suffix): """Create a test dumper appending a suffix to the bytes representation.""" class TestDumper(Dumper): + oid = TEXT_OID + def dump(self, s): return (s + suffix).encode("ascii") diff --git a/tests/test_cursor.py b/tests/test_cursor.py index 1ebebdd9e..981515858 100644 --- a/tests/test_cursor.py +++ b/tests/test_cursor.py @@ -66,8 +66,7 @@ def test_execute_many_results(conn): def test_execute_sequence(conn): cur = conn.cursor() - cast = "::text" if conn.pgconn.server_version < 100000 else "" - rv = cur.execute(f"select %s, %s, %s{cast}", [1, "foo", None]) + rv = cur.execute("select %s::int, %s::text, %s::text", [1, "foo", None]) assert rv is cur assert len(cur._results) == 1 assert cur.pgresult.get_value(0, 0) == b"1" @@ -87,8 +86,7 @@ def test_execute_empty_query(conn, query): def test_fetchone(conn): cur = conn.cursor() - cast = "::text" if conn.pgconn.server_version < 100000 else "" - cur.execute(f"select %s, %s, %s{cast}", [1, "foo", None]) + cur.execute("select %s::int, %s::text, %s::text", [1, "foo", None]) assert cur.pgresult.fformat(0) == 0 row = cur.fetchone() @@ -101,8 +99,7 @@ def test_fetchone(conn): def test_execute_binary_result(conn): cur = conn.cursor(format=psycopg3.pq.Format.BINARY) - cast = "::text" if conn.pgconn.server_version < 100000 else "" - cur.execute(f"select %s, %s{cast}", ["foo", None]) + cur.execute("select %s::text, %s::text", ["foo", None]) assert cur.pgresult.fformat(0) == 1 row = cur.fetchone() diff --git a/tests/test_cursor_async.py b/tests/test_cursor_async.py index 8aca63855..36e8b5267 100644 --- a/tests/test_cursor_async.py +++ b/tests/test_cursor_async.py @@ -69,8 +69,9 @@ async def test_execute_many_results(aconn): async def test_execute_sequence(aconn): cur = aconn.cursor() - cast = "::text" if aconn.pgconn.server_version < 100000 else "" - rv = await cur.execute(f"select %s, %s, %s{cast}", [1, "foo", None]) + rv = await cur.execute( + "select %s::int, %s::text, %s::text", [1, "foo", None] + ) assert rv is cur assert len(cur._results) == 1 assert cur.pgresult.get_value(0, 0) == b"1" @@ -90,8 +91,7 @@ async def test_execute_empty_query(aconn, query): async def test_fetchone(aconn): cur = aconn.cursor() - cast = "::text" if aconn.pgconn.server_version < 100000 else "" - await cur.execute(f"select %s, %s, %s{cast}", [1, "foo", None]) + await cur.execute("select %s::int, %s::text, %s::text", [1, "foo", None]) assert cur.pgresult.fformat(0) == 0 row = await cur.fetchone() @@ -104,8 +104,7 @@ async def test_fetchone(aconn): async def test_execute_binary_result(aconn): cur = aconn.cursor(format=psycopg3.pq.Format.BINARY) - cast = "::text" if aconn.pgconn.server_version < 100000 else "" - await cur.execute(f"select %s, %s{cast}", ["foo", None]) + await cur.execute("select %s::text, %s::text", ["foo", None]) assert cur.pgresult.fformat(0) == 1 row = await cur.fetchone() diff --git a/tests/types/test_composite.py b/tests/types/test_composite.py index 0aa36b798..ca71a7d39 100644 --- a/tests/types/test_composite.py +++ b/tests/types/test_composite.py @@ -57,7 +57,7 @@ def test_load_all_chars(conn, fmt_out): assert res == tuple(map(chr, range(1, 256))) s = "".join(map(chr, range(1, 256))) - res = cur.execute("select row(%s)", [s]).fetchone()[0] + res = cur.execute("select row(%s::text)", [s]).fetchone()[0] assert res == (s,) diff --git a/tests/types/test_text.py b/tests/types/test_text.py index 3c5f47ac3..01b948040 100644 --- a/tests/types/test_text.py +++ b/tests/types/test_text.py @@ -37,7 +37,7 @@ def test_dump_zero(conn, fmt_in): ph = "%s" if fmt_in == Format.TEXT else "%b" s = "foo\x00bar" with pytest.raises(psycopg3.DataError): - cur.execute(f"select {ph}", (s,)) + cur.execute(f"select {ph}::text", (s,)) def test_quote_zero(conn): @@ -76,22 +76,12 @@ def test_load_1char(conn, typename, fmt_out): @pytest.mark.parametrize("fmt_in", [Format.TEXT, Format.BINARY]) -@pytest.mark.parametrize("encoding", ["utf8", "latin9"]) +@pytest.mark.parametrize("encoding", ["utf8", "latin9", "sql_ascii"]) def test_dump_enc(conn, fmt_in, encoding): cur = conn.cursor() ph = "%s" if fmt_in == Format.TEXT else "%b" conn.client_encoding = encoding - (res,) = cur.execute(f"select {ph}::bytea", (eur,)).fetchone() - assert res == eur.encode("utf8") - - -@pytest.mark.parametrize("fmt_in", [Format.TEXT, Format.BINARY]) -def test_dump_ascii(conn, fmt_in): - cur = conn.cursor() - ph = "%s" if fmt_in == Format.TEXT else "%b" - - conn.client_encoding = "sql_ascii" (res,) = cur.execute(f"select ascii({ph})", (eur,)).fetchone() assert res == ord(eur)