Register the C implementation on all the types where it is useful.
return obj.encode(self._encoding)
+@Loader.text(INVALID_OID)
+@Loader.text(builtins["bpchar"].oid)
+@Loader.text(builtins["name"].oid)
@Loader.text(builtins["text"].oid)
-@Loader.binary(builtins["text"].oid)
@Loader.text(builtins["varchar"].oid)
+@Loader.binary(builtins["bpchar"].oid)
+@Loader.binary(builtins["name"].oid)
+@Loader.binary(builtins["text"].oid)
@Loader.binary(builtins["varchar"].oid)
-@Loader.text(INVALID_OID)
class TextLoader(Loader):
_encoding = "utf-8"
return data
-@Loader.text(builtins["name"].oid)
-@Loader.binary(builtins["name"].oid)
-@Loader.text(builtins["bpchar"].oid)
-@Loader.binary(builtins["bpchar"].oid)
-class UnknownLoader(Loader):
-
- _encoding = "utf-8"
-
- def __init__(self, oid: int, context: AdaptContext):
- super().__init__(oid, context)
- conn = self.connection
- if conn:
- self._encoding = conn.client_encoding
-
- def load(self, data: bytes) -> str:
- return data.decode(self._encoding)
-
-
@Dumper.text(bytes)
@Dumper.text(bytearray)
@Dumper.text(memoryview)
StringBinaryDumper.register(str, format=Format.BINARY)
TextLoader.register(oids.INVALID_OID)
+ TextLoader.register(oids.BPCHAR_OID)
+ TextLoader.register(oids.BPCHAR_OID, format=Format.BINARY)
+ TextLoader.register(oids.NAME_OID)
+ TextLoader.register(oids.NAME_OID, format=Format.BINARY)
TextLoader.register(oids.TEXT_OID)
TextLoader.register(oids.TEXT_OID, format=Format.BINARY)
TextLoader.register(oids.VARCHAR_OID)
ByteaLoader.register(oids.BYTEA_OID)
ByteaBinaryLoader.register(oids.BYTEA_OID, format=Format.BINARY)
+ ByteaBinaryLoader.register(oids.INVALID_OID, format=Format.BINARY)
@pytest.mark.parametrize("fmt_out", [Format.TEXT, Format.BINARY])
-@pytest.mark.parametrize("typename", ["text", "varchar"])
+@pytest.mark.parametrize("typename", ["text", "varchar", "name", "bpchar"])
def test_load_ascii(conn, typename, fmt_out):
cur = conn.cursor(format=fmt_out)
assert res == eur.encode("utf8")
-@pytest.mark.parametrize("fmt_out", [Format.TEXT, Format.BINARY])
-@pytest.mark.parametrize("typename", ["name", "bpchar"])
-def test_load_ascii_encanyway(conn, typename, fmt_out):
- cur = conn.cursor(format=fmt_out)
-
- conn.client_encoding = "ascii"
- (res,) = cur.execute(f"select 'aa'::{typename}").fetchone()
- assert res == "aa"
-
-
@pytest.mark.parametrize("fmt_in", [Format.TEXT, Format.BINARY])
@pytest.mark.parametrize("fmt_out", [Format.TEXT, Format.BINARY])
@pytest.mark.parametrize("typename", ["text", "varchar", "name", "bpchar"])