from ..adapt import Dumper, Loader
_PackInt = Callable[[int], bytes]
+_PackFloat = Callable[[float], bytes]
_UnpackInt = Callable[[bytes], Tuple[int]]
_UnpackFloat = Callable[[bytes], Tuple[float]]
_pack_int4 = cast(_PackInt, struct.Struct("!i").pack)
_pack_uint4 = cast(_PackInt, struct.Struct("!I").pack)
_pack_int8 = cast(_PackInt, struct.Struct("!q").pack)
+_pack_float8 = cast(_PackFloat, struct.Struct("!d").pack)
_unpack_int2 = cast(_UnpackInt, struct.Struct("!h").unpack)
_unpack_int4 = cast(_UnpackInt, struct.Struct("!i").unpack)
_unpack_uint4 = cast(_UnpackInt, struct.Struct("!I").unpack)
if value in self._special:
return self._special[value]
- return b" " + value if value.startswith(b"-") else value
+ return value if obj >= 0 else b" " + value
@Dumper.text(int)
class IntDumper(NumberDumper):
- # We don't know the size of it, so we have to return a type big enough
- oid = builtins["numeric"].oid
+ oid = builtins["int8"].oid
+
+
+@Dumper.binary(int)
+class IntBinaryDumper(IntDumper):
+ def dump(self, obj: int) -> bytes:
+ return _pack_int8(obj)
@Dumper.text(float)
class FloatDumper(NumberDumper):
-
oid = builtins["float8"].oid
_special = {
}
+@Dumper.binary(float)
+class FloatBinaryDumper(NumberDumper):
+ def dump(self, obj: float) -> bytes:
+ return _pack_float8(obj)
+
+
@Dumper.text(Decimal)
class DecimalDumper(NumberDumper):
-
oid = builtins["numeric"].oid
_special = {
(int(-(2 ** 63)), "'-9223372036854775808'::bigint"),
],
)
-def test_dump_int(conn, val, expr):
+@pytest.mark.parametrize("fmt_in", [Format.TEXT, Format.BINARY])
+def test_dump_int(conn, val, expr, fmt_in):
+ ph = "%s" if fmt_in == Format.TEXT else "%b"
assert isinstance(val, int)
cur = conn.cursor()
- cur.execute(f"select {expr} = %s", (val,))
+ cur.execute(f"select {expr} = {ph}", (val,))
assert cur.fetchone()[0] is True
assert cur.fetchone() == (val, -val)
-@pytest.mark.xfail
-def test_dump_int_binary():
- # TODO: int binary adaptation (must choose the fitting int2,4,8)
- tx = Transformer()
- n = 1
- tx.get_dumper(n, Format.BINARY).dump(n)
-
-
@pytest.mark.parametrize(
"val, pgtype, want",
[
(float("-inf"), "'-Infinity'"),
],
)
-def test_dump_float(conn, val, expr):
+@pytest.mark.parametrize("fmt_in", [Format.TEXT, Format.BINARY])
+def test_dump_float(conn, val, expr, fmt_in):
+ ph = "%s" if fmt_in == Format.TEXT else "%b"
assert isinstance(val, float)
cur = conn.cursor()
- cur.execute(f"select %s = {expr}::float8", (val,))
+ cur.execute(f"select {ph} = {expr}::float8", (val,))
assert cur.fetchone()[0] is True
assert cur.fetchone()[0] is True
-@pytest.mark.xfail
-def test_dump_float_binary():
- # TODO: float binary adaptation
- tx = Transformer()
- n = 1.0
- tx.get_dumper(n, Format.BINARY).dump(n)
-
-
@pytest.mark.parametrize(
"val, pgtype, want",
[