Behaviour changed in Python 3.11.
# 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).
- cur.execute(f"select %{fmt_in}, %{fmt_in}", ["hello", 10])
+ cur.execute(f"select %{fmt_in.value}, %{fmt_in.value}", ["hello", 10])
assert cur.fetchone() == ("hello", 10)
cur.execute("create table testjson(data jsonb)")
if fmt_in != PyFormat.BINARY:
- cur.execute(f"insert into testjson (data) values (%{fmt_in})", ["{}"])
+ cur.execute(f"insert into testjson (data) values (%{fmt_in.value})", ["{}"])
assert cur.execute("select data from testjson").fetchone() == ({},)
else:
# Binary types cannot be passed as unknown oids.
with pytest.raises(e.DatatypeMismatch):
- cur.execute(f"insert into testjson (data) values (%{fmt_in})", ["{}"])
+ cur.execute(f"insert into testjson (data) values (%{fmt_in.value})", ["{}"])
-@pytest.mark.parametrize("fmt_in", PyFormat.AUTO)
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_no_cast_needed(conn, fmt_in):
# Verify that there is no need of cast in certain common scenario
- cur = conn.execute("select '2021-01-01'::date + %s", [3])
+ cur = conn.execute(f"select '2021-01-01'::date + %{fmt_in.value}", [3])
assert cur.fetchone()[0] == dt.date(2021, 1, 4)
- cur = conn.execute("select '[10, 20, 30]'::jsonb -> %s", [1])
+ cur = conn.execute(f"select '[10, 20, 30]'::jsonb -> %{fmt_in.value}", [1])
assert cur.fetchone()[0] == 20
cur = conn.cursor()
cur.execute("create table testmany (a bigint, b bigint)")
cur.executemany(
- f"insert into testmany values (%{fmt_in}, %{fmt_in})",
+ f"insert into testmany values (%{fmt_in.value}, %{fmt_in.value})",
[[1, None], [3, 4]],
)
with pytest.raises((psycopg.DataError, psycopg.ProgrammingError)):
cur.executemany(
- f"insert into testmany values (%{fmt_in}, %{fmt_in})",
+ f"insert into testmany values (%{fmt_in.value}, %{fmt_in.value})",
[[1, ""], [3, 4]],
)
cur = aconn.cursor()
await cur.execute("create table testmany (a bigint, b bigint)")
await cur.executemany(
- f"insert into testmany values (%{fmt_in}, %{fmt_in})",
+ f"insert into testmany values (%{fmt_in.value}, %{fmt_in.value})",
[[1, None], [3, 4]],
)
with pytest.raises((psycopg.DataError, psycopg.ProgrammingError)):
await cur.executemany(
- f"insert into testmany values (%{fmt_in}, %{fmt_in})",
+ f"insert into testmany values (%{fmt_in.value}, %{fmt_in.value})",
[[1, ""], [3, 4]],
)
@pytest.mark.parametrize("format", PyFormat)
def test_as_string(self, conn, format):
ph = sql.Placeholder(format=format)
- assert ph.as_string(conn) == f"%{format}"
+ assert ph.as_string(conn) == f"%{format.value}"
ph = sql.Placeholder(name="foo", format=format)
- assert ph.as_string(conn) == f"%(foo){format}"
+ assert ph.as_string(conn) == f"%(foo){format.value}"
@pytest.mark.parametrize("format", PyFormat)
def test_as_bytes(self, conn, format):
ph = sql.Placeholder(format=format)
- assert ph.as_bytes(conn) == f"%{format}".encode("ascii")
+ assert ph.as_bytes(conn) == f"%{format.value}".encode("ascii")
ph = sql.Placeholder(name="foo", format=format)
- assert ph.as_bytes(conn) == f"%(foo){format}".encode("ascii")
+ assert ph.as_bytes(conn) == f"%(foo){format.value}".encode("ascii")
class TestValues:
@pytest.mark.parametrize("obj, want", tests_str)
def test_dump_list_str(conn, obj, want, fmt_in):
cur = conn.cursor()
- cur.execute(f"select %{fmt_in}::text[] = %s::text[]", (obj, want))
+ cur.execute(f"select %{fmt_in.value}::text[] = %s::text[]", (obj, want))
assert cur.fetchone()[0]
cur = conn.cursor(binary=fmt_out)
for i in range(1, 256):
c = chr(i)
- cur.execute(f"select %{fmt_in}::text[]", ([c],))
+ cur.execute(f"select %{fmt_in.value}::text[]", ([c],))
assert cur.fetchone()[0] == [c]
a = list(map(chr, range(1, 256)))
a.append("\u20ac")
- cur.execute(f"select %{fmt_in}::text[]", (a,))
+ cur.execute(f"select %{fmt_in.value}::text[]", (a,))
assert cur.fetchone()[0] == a
s = "".join(a)
- cur.execute(f"select %{fmt_in}::text[]", ([s],))
+ cur.execute(f"select %{fmt_in.value}::text[]", ([s],))
assert cur.fetchone()[0] == [s]
obj = [wrapper(1), wrapper(0), wrapper(-1), None]
cur = conn.cursor(binary=fmt_out)
- got = cur.execute(f"select %{fmt_in}", [obj]).fetchone()[0]
+ got = cur.execute(f"select %{fmt_in.value}", [obj]).fetchone()[0]
assert got == obj
for i in got:
if i is not None:
conn.execute("create table testarrays (col1 bigint[], col2 bigint[])")
# pro tip: don't get confused with the types
f1, f2 = conn.execute(
- f"insert into testarrays values (%{fmt_in}, %{fmt_in}) returning *",
+ f"insert into testarrays values (%{fmt_in.value}, %{fmt_in.value}) returning *",
(objs, []),
).fetchone()
assert f1 == objs
cur = conn.cursor()
cur.execute("create table test (id serial primary key, data date[])")
with conn.transaction():
- cur.execute(f"insert into test (data) values (%{fmt_in})", ([],))
+ cur.execute(f"insert into test (data) values (%{fmt_in.value})", ([],))
cur.execute("select data from test")
assert cur.fetchone() == ([],)
# test untyped list in a filter
- cur.execute(f"select data from test where id = any(%{fmt_in})", ([1],))
+ cur.execute(f"select data from test where id = any(%{fmt_in.value})", ([1],))
assert cur.fetchone()
- cur.execute(f"select data from test where id = any(%{fmt_in})", ([],))
+ cur.execute(f"select data from test where id = any(%{fmt_in.value})", ([],))
assert not cur.fetchone()
def test_empty_list_after_choice(conn, fmt_in):
cur = conn.cursor()
cur.execute("create table test (id serial primary key, data float[])")
- cur.executemany(f"insert into test (data) values (%{fmt_in})", [([1.0],), ([],)])
+ cur.executemany(
+ f"insert into test (data) values (%{fmt_in.value})", [([1.0],), ([],)]
+ )
cur.execute("select data from test order by id")
assert cur.fetchall() == [([1.0],), ([],)]
@pytest.mark.parametrize("b", [True, False])
def test_roundtrip_bool(conn, b, fmt_in, fmt_out):
cur = conn.cursor(binary=fmt_out)
- result = cur.execute(f"select %{fmt_in}", (b,)).fetchone()[0]
+ result = cur.execute(f"select %{fmt_in.value}", (b,)).fetchone()[0]
assert cur.pgresult.fformat(0) == fmt_out
if b is not None:
assert cur.pgresult.ftype(0) == builtins["bool"].oid
assert result is b
- result = cur.execute(f"select %{fmt_in}", ([b],)).fetchone()[0]
+ result = cur.execute(f"select %{fmt_in.value}", ([b],)).fetchone()[0]
assert cur.pgresult.fformat(0) == fmt_out
if b is not None:
assert cur.pgresult.ftype(0) == builtins["bool"].array_oid
register_composite(info, conn)
cur = conn.execute(
- f"select pg_typeof(%{fmt_in})",
+ f"select pg_typeof(%{fmt_in.value})",
[info.python_type(10, Range(empty=True), [])],
)
assert cur.fetchone()[0] == "tmptype"
cur = conn.cursor()
for i in range(1, 256):
(res,) = cur.execute(
- f"select row(chr(%s::int), 1, 1.0)::testcomp = %{fmt_in}::testcomp",
+ f"select row(chr(%s::int), 1, 1.0)::testcomp = %{fmt_in.value}::testcomp",
(i, (chr(i), 1, 1.0)),
).fetchone()
assert res is True
for i in range(1, 256):
obj = factory(chr(i), 1, 1.0)
(res,) = cur.execute(
- f"select row(chr(%s::int), 1, 1.0)::testcomp = %{fmt_in}", (i, obj)
+ f"select row(chr(%s::int), 1, 1.0)::testcomp = %{fmt_in.value}", (i, obj)
).fetchone()
assert res is True
obj = factory("foo", 1, None)
rec = cur.execute(
f"""
- select row('foo', 1, NULL)::testcomp = %(obj){fmt_in}, %(obj){fmt_in}::text
+ select row('foo', 1, NULL)::testcomp = %(obj){fmt_in.value},
+ %(obj){fmt_in.value}::text
""",
{"obj": obj},
).fetchone()
def test_dump_date(self, conn, val, expr, fmt_in):
val = as_date(val)
cur = conn.cursor()
- cur.execute(f"select '{expr}'::date = %{fmt_in}", (val,))
+ cur.execute(f"select '{expr}'::date = %{fmt_in.value}", (val,))
assert cur.fetchone()[0] is True
cur.execute(
def test_dump_datetime(self, conn, val, expr, fmt_in):
cur = conn.cursor()
cur.execute("set timezone to '+02:00'")
- cur.execute(f"select %{fmt_in}", (as_dt(val),))
- cur.execute(f"select '{expr}'::timestamp = %{fmt_in}", (as_dt(val),))
+ cur.execute(f"select %{fmt_in.value}", (as_dt(val),))
+ cur.execute(f"select '{expr}'::timestamp = %{fmt_in.value}", (as_dt(val),))
cur.execute(
f"""
- select '{expr}'::timestamp = %(val){fmt_in},
- '{expr}', %(val){fmt_in}::text
+ select '{expr}'::timestamp = %(val){fmt_in.value},
+ '{expr}', %(val){fmt_in.value}::text
""",
{"val": as_dt(val)},
)
cur.execute("set timezone to '-02:00'")
cur.execute(
f"""
- select '{expr}'::timestamptz = %(val){fmt_in},
- '{expr}', %(val){fmt_in}::text
+ select '{expr}'::timestamptz = %(val){fmt_in.value},
+ '{expr}', %(val){fmt_in.value}::text
""",
{"val": as_dt(val)},
)
val = as_dt(val)
cur = conn.cursor()
cur.execute(
- f"select pg_typeof(%{fmt_in}) = %s::regtype, %{fmt_in}",
+ f"select pg_typeof(%{fmt_in.value}) = %s::regtype, %{fmt_in.value}",
[val, type, val],
)
rec = cur.fetchone()
cur = conn.cursor()
cur.execute(
f"""
- select '{expr}'::time = %(val){fmt_in},
- '{expr}'::time::text, %(val){fmt_in}::text
+ select '{expr}'::time = %(val){fmt_in.value},
+ '{expr}'::time::text, %(val){fmt_in.value}::text
""",
{"val": as_time(val)},
)
def test_dump_timetz(self, conn, val, expr, fmt_in):
cur = conn.cursor()
cur.execute("set timezone to '-02:00'")
- cur.execute(f"select '{expr}'::timetz = %{fmt_in}", (as_time(val),))
+ cur.execute(f"select '{expr}'::timetz = %{fmt_in.value}", (as_time(val),))
assert cur.fetchone()[0] is True
@pytest.mark.parametrize(
val = as_time(val)
cur = conn.cursor()
cur.execute(
- f"select pg_typeof(%{fmt_in}) = %s::regtype, %{fmt_in}",
+ f"select pg_typeof(%{fmt_in.value}) = %s::regtype, %{fmt_in.value}",
[val, type, val],
)
rec = cur.fetchone()
def test_json_dump(conn, val, fmt_in):
obj = json.loads(val)
cur = conn.cursor()
- cur.execute(f"select pg_typeof(%{fmt_in}) = 'json'::regtype", (Json(obj),))
+ cur.execute(f"select pg_typeof(%{fmt_in.value}) = 'json'::regtype", (Json(obj),))
assert cur.fetchone()[0] is True
- cur.execute(f"select %{fmt_in}::text = %s::json::text", (Json(obj), val))
+ cur.execute(f"select %{fmt_in.value}::text = %s::json::text", (Json(obj), val))
assert cur.fetchone()[0] is True
def test_jsonb_dump(conn, val, fmt_in):
obj = json.loads(val)
cur = conn.cursor()
- cur.execute(f"select %{fmt_in} = %s::jsonb", (Jsonb(obj), val))
+ cur.execute(f"select %{fmt_in.value} = %s::jsonb", (Jsonb(obj), val))
assert cur.fetchone()[0] is True
set_json_dumps(my_dumps)
try:
- cur.execute(f"select %{fmt_in}->>'baz' = 'qux'", (wrapper(obj),))
+ cur.execute(f"select %{fmt_in.value}->>'baz' = 'qux'", (wrapper(obj),))
assert cur.fetchone()[0] is True
finally:
set_json_dumps(json.dumps)
cur2 = conn.cursor()
set_json_dumps(my_dumps, cur2)
- cur1.execute(f"select %{fmt_in}->>'baz'", (wrapper(obj),))
+ cur1.execute(f"select %{fmt_in.value}->>'baz'", (wrapper(obj),))
assert cur1.fetchone()[0] is None
- cur2.execute(f"select %{fmt_in}->>'baz'", (wrapper(obj),))
+ cur2.execute(f"select %{fmt_in.value}->>'baz'", (wrapper(obj),))
assert cur2.fetchone()[0] == "qux"
wrapper = getattr(psycopg.types.json, wrapper)
obj = {"foo": "bar"}
cur = conn.cursor()
- cur.execute(f"select %{fmt_in}->>'baz' = 'qux'", (wrapper(obj, my_dumps),))
+ cur.execute(f"select %{fmt_in.value}->>'baz' = 'qux'", (wrapper(obj, my_dumps),))
assert cur.fetchone()[0] is True
@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_builtin_empty(conn, pgtype, fmt_in):
mr = Multirange() # type: ignore[var-annotated]
- cur = conn.execute(f"select '{{}}'::{pgtype} = %{fmt_in}", (mr,))
+ cur = conn.execute(f"select '{{}}'::{pgtype} = %{fmt_in.value}", (mr,))
assert cur.fetchone()[0] is True
mr = wrapper()
rec = conn.execute(
f"""
- select '{{}}' = %(mr){fmt_in},
- %(mr){fmt_in}::text,
- pg_typeof(%(mr){fmt_in})::oid
+ select '{{}}' = %(mr){fmt_in.value},
+ %(mr){fmt_in.value}::text,
+ pg_typeof(%(mr){fmt_in.value})::oid
""",
{"mr": mr},
).fetchone()
mr1 = Multirange() # type: ignore[var-annotated]
mr2 = Multirange([Range(bounds="()")]) # type: ignore[var-annotated]
cur = conn.execute(
- f"select array['{{}}'::{pgtype}, '{{(,)}}'::{pgtype}] = %{fmt_in}",
+ f"select array['{{}}'::{pgtype}, '{{(,)}}'::{pgtype}] = %{fmt_in.value}",
([mr1, mr2],),
)
assert cur.fetchone()[0] is True
mr2 = Multirange([Range(bounds="()")]) # type: ignore[var-annotated]
cur = conn.execute(
f"""
- select array['{{}}'::{pgtype}, '{{(,)}}'::{pgtype}] = %{fmt_in}::{pgtype}[]
+ select array['{{}}'::{pgtype},
+ '{{(,)}}'::{pgtype}] = %{fmt_in.value}::{pgtype}[]
""",
([mr1, mr2],),
)
wrapper = getattr(multirange, wrapper)
mr1 = Multirange() # type: ignore[var-annotated]
mr2 = Multirange([Range(bounds="()")]) # type: ignore[var-annotated]
- cur = conn.execute(f"""select '{{"{{}}","{{(,)}}"}}' = %{fmt_in}""", ([mr1, mr2],))
+ cur = conn.execute(
+ f"""select '{{"{{}}","{{(,)}}"}}' = %{fmt_in.value}""", ([mr1, mr2],)
+ )
assert cur.fetchone()[0] is True
mr = Multirange(ranges)
rname = pgtype.replace("multi", "")
phs = ", ".join([f"%s::{rname}"] * len(ranges))
- cur = conn.execute(f"select {pgtype}({phs}) = %{fmt_in}", ranges + [mr])
+ cur = conn.execute(f"select {pgtype}({phs}) = %{fmt_in.value}", ranges + [mr])
assert cur.fetchone()[0] is True
@pytest.mark.parametrize("val", ["192.168.0.1", "2001:db8::"])
def test_address_dump(conn, fmt_in, val):
cur = conn.cursor()
- cur.execute(f"select %{fmt_in} = %s::inet", (ipaddress.ip_address(val), val))
+ cur.execute(f"select %{fmt_in.value} = %s::inet", (ipaddress.ip_address(val), val))
assert cur.fetchone()[0] is True
cur.execute(
- f"select %{fmt_in} = array[null, %s]::inet[]",
+ f"select %{fmt_in.value} = array[null, %s]::inet[]",
([None, ipaddress.ip_interface(val)], val),
)
assert cur.fetchone()[0] is True
def test_interface_dump(conn, fmt_in, val):
cur = conn.cursor()
rec = cur.execute(
- f"select %(val){fmt_in} = %(repr)s::inet, %(val){fmt_in}, %(repr)s::inet",
+ f"select %(val){fmt_in.value} = %(repr)s::inet,"
+ f" %(val){fmt_in.value}, %(repr)s::inet",
{"val": ipaddress.ip_interface(val), "repr": val},
).fetchone()
assert rec[0] is True, f"{rec[1]} != {rec[2]}"
cur.execute(
- f"select %{fmt_in} = array[null, %s]::inet[]",
+ f"select %{fmt_in.value} = array[null, %s]::inet[]",
([None, ipaddress.ip_interface(val)], val),
)
assert cur.fetchone()[0] is True
@pytest.mark.parametrize("val", ["127.0.0.0/24", "::ffff:102:300/128"])
def test_network_dump(conn, fmt_in, val):
cur = conn.cursor()
- cur.execute(f"select %{fmt_in} = %s::cidr", (ipaddress.ip_network(val), val))
+ cur.execute(f"select %{fmt_in.value} = %s::cidr", (ipaddress.ip_network(val), val))
assert cur.fetchone()[0] is True
cur.execute(
- f"select %{fmt_in} = array[NULL, %s]::cidr[]",
+ f"select %{fmt_in.value} = array[NULL, %s]::cidr[]",
([None, ipaddress.ip_network(val)], val),
)
assert cur.fetchone()[0] is True
ipaddress.IPv6Network("::1/128"),
]
cur = conn.cursor()
- cur.execute(f"select %{fmt_in}", (val,))
+ cur.execute(f"select %{fmt_in.value}", (val,))
got = cur.fetchone()[0]
assert val == got
def test_dump_int(conn, val, expr, fmt_in):
assert isinstance(val, int)
cur = conn.cursor()
- cur.execute(f"select {expr} = %{fmt_in}", (val,))
+ cur.execute(f"select {expr} = %{fmt_in.value}", (val,))
assert cur.fetchone()[0] is True
@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_int_subtypes(conn, val, expr, fmt_in):
cur = conn.cursor()
- cur.execute(f"select pg_typeof({expr}) = pg_typeof(%{fmt_in})", (val,))
+ cur.execute(f"select pg_typeof({expr}) = pg_typeof(%{fmt_in.value})", (val,))
assert cur.fetchone()[0] is True
cur.execute(
- f"select {expr} = %(v){fmt_in}, {expr}::text, %(v){fmt_in}::text",
+ f"select {expr} = %(v){fmt_in.value}, {expr}::text, %(v){fmt_in.value}::text",
{"v": val},
)
ok, want, got = cur.fetchone()
@pytest.mark.parametrize("enum", [MyEnum, MyMixinEnum])
def test_dump_enum(conn, fmt_in, enum):
cur = conn.cursor()
- cur.execute(f"select %{fmt_in}", (enum.foo,))
+ cur.execute(f"select %{fmt_in.value}", (enum.foo,))
(res,) = cur.fetchone()
assert res == enum.foo.value
def test_dump_float(conn, val, expr, fmt_in):
assert isinstance(val, float)
cur = conn.cursor()
- cur.execute(f"select %{fmt_in} = {expr}::float8", (val,))
+ cur.execute(f"select %{fmt_in.value} = {expr}::float8", (val,))
assert cur.fetchone()[0] is True
def test_roundtrip_numeric(conn, val, fmt_in, fmt_out):
cur = conn.cursor(binary=fmt_out)
val = Decimal(val)
- cur.execute(f"select %{fmt_in}", (val,))
+ cur.execute(f"select %{fmt_in.value}", (val,))
result = cur.fetchone()[0]
assert isinstance(result, Decimal)
if val.is_nan():
for f in funcs:
expr = f(i)
val = Decimal(expr)
- cur.execute(f"select %{fmt_in}::text, %s::decimal::text", [val, expr])
+ cur.execute(f"select %{fmt_in.value}::text, %s::decimal::text", [val, expr])
want, got = cur.fetchone()
assert got == want
@pytest.mark.parametrize("fmt_in", PyFormat)
def test_repr_wrapper(conn, wrapper, fmt_in):
wrapper = getattr(psycopg.types.numeric, wrapper)
- cur = conn.execute(f"select pg_typeof(%{fmt_in})::oid", [wrapper(0)])
+ cur = conn.execute(f"select pg_typeof(%{fmt_in.value})::oid", [wrapper(0)])
oid = cur.fetchone()[0]
assert oid == psycopg.postgres.types[wrapper.__name__.lower()].oid
@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_builtin_empty(conn, pgtype, fmt_in):
r = Range(empty=True) # type: ignore[var-annotated]
- cur = conn.execute(f"select 'empty'::{pgtype} = %{fmt_in}", (r,))
+ cur = conn.execute(f"select 'empty'::{pgtype} = %{fmt_in.value}", (r,))
assert cur.fetchone()[0] is True
def test_dump_builtin_empty_wrapper(conn, wrapper, fmt_in):
wrapper = getattr(range_module, wrapper)
r = wrapper(empty=True)
- cur = conn.execute(f"select 'empty' = %{fmt_in}", (r,))
+ cur = conn.execute(f"select 'empty' = %{fmt_in.value}", (r,))
assert cur.fetchone()[0] is True
r1 = Range(empty=True) # type: ignore[var-annotated]
r2 = Range(bounds="()") # type: ignore[var-annotated]
cur = conn.execute(
- f"select array['empty'::{pgtype}, '(,)'::{pgtype}] = %{fmt_in}",
+ f"select array['empty'::{pgtype}, '(,)'::{pgtype}] = %{fmt_in.value}",
([r1, r2],),
)
assert cur.fetchone()[0] is True
r1 = Range(empty=True) # type: ignore[var-annotated]
r2 = Range(bounds="()") # type: ignore[var-annotated]
cur = conn.execute(
- f"select array['empty'::{pgtype}, '(,)'::{pgtype}] = %{fmt_in}::{pgtype}[]",
+ f"select array['empty'::{pgtype}, '(,)'::{pgtype}] "
+ f"= %{fmt_in.value}::{pgtype}[]",
([r1, r2],),
)
assert cur.fetchone()[0] is True
wrapper = getattr(range_module, wrapper)
r1 = wrapper(empty=True)
r2 = wrapper(bounds="()")
- cur = conn.execute(f"""select '{{empty,"(,)"}}' = %{fmt_in}""", ([r1, r2],))
+ cur = conn.execute(f"""select '{{empty,"(,)"}}' = %{fmt_in.value}""", ([r1, r2],))
assert cur.fetchone()[0] is True
r = Range(min, max, bounds) # type: ignore[var-annotated]
sub = type2sub[pgtype]
cur = conn.execute(
- f"select {pgtype}(%s::{sub}, %s::{sub}, %s) = %{fmt_in}",
+ f"select {pgtype}(%s::{sub}, %s::{sub}, %s) = %{fmt_in.value}",
(min, max, bounds, r),
)
assert cur.fetchone()[0] is True
def test_dump_1char(conn, fmt_in):
cur = conn.cursor()
for i in range(1, 256):
- cur.execute(f"select %{fmt_in} = chr(%s)", (chr(i), i))
+ cur.execute(f"select %{fmt_in.value} = chr(%s)", (chr(i), i))
assert cur.fetchone()[0] is True, chr(i)
cur = conn.cursor()
s = "foo\x00bar"
with pytest.raises(psycopg.DataError):
- cur.execute(f"select %{fmt_in}::text", (s,))
+ cur.execute(f"select %{fmt_in.value}::text", (s,))
def test_quote_zero(conn):
cur = conn.cursor()
conn.execute(f"set client_encoding to {encoding}")
- (res,) = cur.execute(f"select ascii(%{fmt_in})", (eur,)).fetchone()
+ (res,) = cur.execute(f"select ascii(%{fmt_in.value})", (eur,)).fetchone()
assert res == ord(eur)
conn.execute("set client_encoding to latin1")
with pytest.raises(UnicodeEncodeError):
- cur.execute(f"select %{fmt_in}::bytea", (eur,))
+ cur.execute(f"select %{fmt_in.value}::bytea", (eur,))
@pytest.mark.parametrize("fmt_in", PyFormat)
conn.execute("set client_encoding to utf8")
with pytest.raises(UnicodeEncodeError):
- cur.execute(f"select %{fmt_in}", ("\uddf8",))
+ cur.execute(f"select %{fmt_in.value}", ("\uddf8",))
@pytest.mark.parametrize("fmt_in", [PyFormat.AUTO, PyFormat.TEXT])
cur = conn.cursor()
cur.execute("create type myenum as enum ('foo', 'bar')")
cur.execute("create table with_enum (e myenum)")
- cur.execute(f"insert into with_enum (e) values (%{fmt_in})", (MyEnum.foo,))
+ cur.execute(f"insert into with_enum (e) values (%{fmt_in.value})", (MyEnum.foo,))
(res,) = cur.execute("select e from with_enum").fetchone()
assert res == "foo"
conn.autocommit = True
with pytest.raises(e.IndeterminateDatatype):
- conn.execute(f"select concat(%{fmt_in}, %{fmt_in})", ["foo", "bar"])
+ conn.execute(f"select concat(%{fmt_in.value}, %{fmt_in.value})", ["foo", "bar"])
conn.adapters.register_dumper(str, psycopg.types.string.StrDumper)
- cur = conn.execute(f"select concat(%{fmt_in}, %{fmt_in})", ["foo", "bar"])
+ cur = conn.execute(
+ f"select concat(%{fmt_in.value}, %{fmt_in.value})", ["foo", "bar"]
+ )
assert cur.fetchone()[0] == "foobar"
cur = conn.cursor(binary=fmt_out)
a = list(map(chr, range(1, 256))) + [eur]
- (res,) = cur.execute(f"select %{fmt_in}::{typename}[]", (a,)).fetchone()
+ (res,) = cur.execute(f"select %{fmt_in.value}::{typename}[]", (a,)).fetchone()
assert res == a
cur = conn.cursor(binary=fmt_out)
a = list(map(chr, range(1, 256))) + [eur]
exp = [s.encode() for s in a]
- (res,) = cur.execute(f"select %{fmt_in}::text[]", (a,)).fetchone()
+ (res,) = cur.execute(f"select %{fmt_in.value}::text[]", (a,)).fetchone()
assert res == exp
cur = conn.cursor()
for i in range(0, 256):
obj = pytype(bytes([i]))
- cur.execute(f"select %{fmt_in} = set_byte('x', 0, %s)", (obj, i))
+ cur.execute(f"select %{fmt_in.value} = set_byte('x', 0, %s)", (obj, i))
assert cur.fetchone()[0] is True, i
- cur.execute(f"select %{fmt_in} = array[set_byte('x', 0, %s)]", ([obj], i))
+ cur.execute(f"select %{fmt_in.value} = array[set_byte('x', 0, %s)]", ([obj], i))
assert cur.fetchone()[0] is True
def test_bytea_array(conn, fmt_in, fmt_out):
cur = conn.cursor(binary=fmt_out)
a = [bytes(range(0, 256))]
- (res,) = cur.execute(f"select %{fmt_in}::bytea[]", (a,)).fetchone()
+ (res,) = cur.execute(f"select %{fmt_in.value}::bytea[]", (a,)).fetchone()
assert res == a
def test_uuid_dump(conn, fmt_in):
val = "12345678123456781234567812345679"
cur = conn.cursor()
- cur.execute(f"select %{fmt_in} = %s::uuid", (UUID(val), val))
+ cur.execute(f"select %{fmt_in.value} = %s::uuid", (UUID(val), val))
assert cur.fetchone()[0] is True