from .types.oids import builtins
from .adapt import Dumper
-BYTEA_OID = builtins["bytea"].oid
-
class DBAPITypeObject:
def __init__(self, name: str, type_names: Sequence[str]):
@Dumper.text(Binary)
-class TextBinaryDumper(Dumper):
+class BinaryDumper(Dumper):
+
+ oid = builtins["bytea"].oid
+
def dump(self, obj: Binary) -> bytes:
wrapped = obj.obj
if isinstance(wrapped, bytes):
else:
return bytes(wrapped)
- @property
- def oid(self) -> int:
- return BYTEA_OID
-
def Date(year: int, month: int, day: int) -> dt.date:
return dt.date(year, month, day)
@Dumper.text(list)
-class TextListDumper(BaseListDumper):
+class ListDumper(BaseListDumper):
# from https://www.postgresql.org/docs/current/arrays.html#ARRAYS-IO
#
# The array output routine will put double quotes around element values if
@Dumper.binary(list)
-class BinaryListDumper(BaseListDumper):
+class ListBinaryDumper(BaseListDumper):
def dump(self, obj: List[Any]) -> bytes:
if not obj:
return _struct_head.pack(0, 0, TEXT_OID)
self._tx = Transformer(context)
-class TextArrayLoader(BaseArrayLoader):
+class ArrayLoader(BaseArrayLoader):
# Tokenize an array representation into item and brackets
# TODO: currently recognise only , as delimiter. Should be configured
_struct_len = struct.Struct("!i")
-class BinaryArrayLoader(BaseArrayLoader):
+class ArrayBinaryLoader(BaseArrayLoader):
def load(self, data: bytes) -> List[Any]:
ndims, hasnull, oid = _struct_head.unpack_from(data[:12])
if not ndims:
name = f"oid{base_oid}"
for format, base in (
- (Format.TEXT, TextArrayLoader),
- (Format.BINARY, BinaryArrayLoader),
+ (Format.TEXT, ArrayLoader),
+ (Format.BINARY, ArrayBinaryLoader),
):
- lname = f"{name.title()}Array{format.name.title()}Loader"
+ lname = f"{name.title()}Array{'Binary' if format else ''}Loader"
loader: Type[Loader] = type(lname, (base,), {"base_oid": base_oid})
loader.register(array_oid, context=context, format=format)
# generate and register a customized binary loader
loader = type(
- f"Binary{info.name.title()}Loader",
- (BinaryCompositeLoader,),
+ f"{info.name.title()}BinaryLoader",
+ (CompositeBinaryLoader,),
{"factory": factory},
)
loader.register(info.oid, context=context, format=Format.BINARY)
@Dumper.text(tuple)
-class TextTupleDumper(Dumper):
+class TupleDumper(Dumper):
def __init__(self, src: type, context: AdaptContext = None):
super().__init__(src, context)
self._tx = Transformer(context)
@Loader.binary(builtins["record"].oid)
-class BinaryRecordLoader(BaseCompositeLoader):
+class RecordBinaryLoader(BaseCompositeLoader):
_types_set = False
def load(self, data: bytes) -> Tuple[Any, ...]:
)
-class BinaryCompositeLoader(BinaryRecordLoader):
+class CompositeBinaryLoader(RecordBinaryLoader):
factory: Callable[..., Any]
def load(self, data: bytes) -> Any:
pass
-class JsonB(_JsonWrapper):
+class Jsonb(_JsonWrapper):
pass
class _JsonDumper(Dumper):
- _oid: int
-
def dump(
self, obj: _JsonWrapper, __encode: EncodeFunc = _encode_utf8
) -> bytes:
return __encode(obj.dumps())[0]
- @property
- def oid(self) -> int:
- return self._oid
-
@Dumper.text(Json)
@Dumper.binary(Json)
class JsonDumper(_JsonDumper):
- _oid = JSON_OID
+ oid = JSON_OID
-@Dumper.text(JsonB)
-class JsonBDumper(_JsonDumper):
- _oid = JSONB_OID
+@Dumper.text(Jsonb)
+class JsonbDumper(_JsonDumper):
+ oid = JSONB_OID
-@Dumper.binary(JsonB)
-class BinaryJsonBDumper(JsonBDumper):
+@Dumper.binary(Jsonb)
+class JsonbBinaryDumper(JsonbDumper):
def dump(
self, obj: _JsonWrapper, __encode: EncodeFunc = _encode_utf8
) -> bytes:
@Loader.binary(builtins["jsonb"].oid)
-class BinaryJsonBLoader(Loader):
+class JsonbBinaryLoader(Loader):
def load(self, data: bytes) -> Any:
if data and data[0] != 1:
raise DataError("unknown jsonb binary format: {data[0]}")
UnpackInt = Callable[[bytes], Tuple[int]]
UnpackFloat = Callable[[bytes], Tuple[float]]
-FLOAT8_OID = builtins["float8"].oid
-NUMERIC_OID = builtins["numeric"].oid
-
_encode_ascii = codecs.lookup("ascii").encode
_decode_ascii = codecs.lookup("ascii").decode
@Dumper.text(int)
-class TextIntDumper(NumberDumper):
- @property
- def oid(self) -> int:
- # We don't know the size of it, so we have to return a type big enough
- return NUMERIC_OID
+class IntDumper(NumberDumper):
+ # We don't know the size of it, so we have to return a type big enough
+ oid = builtins["numeric"].oid
@Dumper.text(float)
-class TextFloatDumper(NumberDumper):
+class FloatDumper(NumberDumper):
+
+ oid = builtins["float8"].oid
+
_special = {
b"inf": b"'Infinity'::float8",
b"-inf": b"'-Infinity'::float8",
b"nan": b"'NaN'::float8",
}
- @property
- def oid(self) -> int:
- # Float can't be bigger than this instead
- return FLOAT8_OID
-
@Dumper.text(Decimal)
-class TextDecimalDumper(NumberDumper):
+class DecimalDumper(NumberDumper):
+
+ oid = builtins["numeric"].oid
+
_special = {
b"Infinity": b"'Infinity'::numeric",
b"-Infinity": b"'-Infinity'::numeric",
b"NaN": b"'NaN'::numeric",
}
- @property
- def oid(self) -> int:
- return NUMERIC_OID
-
@Loader.text(builtins["int2"].oid)
@Loader.text(builtins["int4"].oid)
@Loader.text(builtins["int8"].oid)
@Loader.text(builtins["oid"].oid)
-class TextIntLoader(Loader):
+class IntLoader(Loader):
def load(self, data: bytes, __decode: DecodeFunc = _decode_ascii) -> int:
return int(__decode(data)[0])
@Loader.binary(builtins["int2"].oid)
-class BinaryInt2Loader(Loader):
+class Int2BinaryLoader(Loader):
def load(
self,
data: bytes,
@Loader.binary(builtins["int4"].oid)
-class BinaryInt4Loader(Loader):
+class Int4BinaryLoader(Loader):
def load(
self,
data: bytes,
@Loader.binary(builtins["int8"].oid)
-class BinaryInt8Loader(Loader):
+class Int8BinaryLoader(Loader):
def load(
self,
data: bytes,
@Loader.binary(builtins["oid"].oid)
-class BinaryOidLoader(Loader):
+class OidBinaryLoader(Loader):
def load(
self,
data: bytes,
@Loader.text(builtins["float4"].oid)
@Loader.text(builtins["float8"].oid)
-class TextFloatLoader(Loader):
+class FloatLoader(Loader):
def load(self, data: bytes) -> float:
# it supports bytes directly
return float(data)
@Loader.binary(builtins["float4"].oid)
-class BinaryFloat4Loader(Loader):
+class Float4BinaryLoader(Loader):
def load(
self,
data: bytes,
@Loader.binary(builtins["float8"].oid)
-class BinaryFloat8Loader(Loader):
+class Float8BinaryLoader(Loader):
def load(
self,
data: bytes,
@Loader.text(builtins["numeric"].oid)
-class TextNumericLoader(Loader):
+class NumericLoader(Loader):
def load(
self, data: bytes, __decode: DecodeFunc = _decode_ascii
) -> Decimal:
@Dumper.text(bool)
-class TextBoolDumper(Dumper):
+class BoolDumper(Dumper):
+
+ oid = BOOL_OID
+
def dump(self, obj: bool) -> bytes:
return b"t" if obj else b"f"
def quote(self, obj: bool) -> bytes:
return b"true" if obj else b"false"
- @property
- def oid(self) -> int:
- return BOOL_OID
-
@Dumper.binary(bool)
class BinaryBoolDumper(Dumper):
+
+ oid = BOOL_OID
+
def dump(self, obj: bool) -> bytes:
return b"\x01" if obj else b"\x00"
- @property
- def oid(self) -> int:
- return BOOL_OID
-
@Dumper.text(type(None))
class NoneDumper(Dumper):
@Loader.text(builtins["bool"].oid)
-class TextBoolLoader(Loader):
+class BoolLoader(Loader):
def load(
self,
data: bytes,
@Loader.text(builtins["varchar"].oid)
@Loader.binary(builtins["varchar"].oid)
@Loader.text(INVALID_OID)
-class StringLoader(Loader):
+class TextLoader(Loader):
decode: Optional[DecodeFunc]
@Dumper.binary(bytes)
-class BinaryBytesDumper(Dumper):
+class BytesBinaryDumper(Dumper):
def dump(self, b: bytes) -> bytes:
return b
@Loader.text(builtins["bytea"].oid)
-class TextByteaLoader(Loader):
+class ByteaLoader(Loader):
_escaping: "EscapingProto"
def __init__(self, oid: int, context: AdaptContext = None):
@Loader.binary(builtins["bytea"].oid)
@Loader.binary(INVALID_OID)
-class BinaryByteaLoader(Loader):
+class ByteaBinaryLoader(Loader):
def load(self, data: bytes) -> bytes:
return data
PyLong_FromLong, PyLong_FromLongLong, PyLong_FromUnsignedLong)
-cdef class TextIntLoader(CLoader):
+cdef class IntLoader(CLoader):
cdef object cload(self, const char *data, size_t length):
return int(data)
-cdef class BinaryInt2Loader(CLoader):
+cdef class Int2BinaryLoader(CLoader):
cdef object cload(self, const char *data, size_t length):
return PyLong_FromLong(<int16_t>be16toh((<uint16_t *>data)[0]))
-cdef class BinaryInt4Loader(CLoader):
+cdef class Int4BinaryLoader(CLoader):
cdef object cload(self, const char *data, size_t length):
return PyLong_FromLong(<int32_t>be32toh((<uint32_t *>data)[0]))
-cdef class BinaryInt8Loader(CLoader):
+cdef class Int8BinaryLoader(CLoader):
cdef object cload(self, const char *data, size_t length):
return PyLong_FromLongLong(<int64_t>be64toh((<uint64_t *>data)[0]))
-cdef class BinaryOidLoader(CLoader):
+cdef class OidBinaryLoader(CLoader):
cdef object cload(self, const char *data, size_t length):
return PyLong_FromUnsignedLong(be32toh((<uint32_t *>data)[0]))
-cdef class BinaryBoolLoader(CLoader):
+cdef class BoolBinaryLoader(CLoader):
cdef object cload(self, const char *data, size_t length):
if data[0]:
return True
from psycopg3.adapt import Loader
from psycopg3.types import builtins
- TextIntLoader.register(builtins["int2"].oid)
- TextIntLoader.register(builtins["int4"].oid)
- TextIntLoader.register(builtins["int8"].oid)
- TextIntLoader.register(builtins["oid"].oid)
+ IntLoader.register(builtins["int2"].oid)
+ IntLoader.register(builtins["int4"].oid)
+ IntLoader.register(builtins["int8"].oid)
+ IntLoader.register(builtins["oid"].oid)
- BinaryInt2Loader.register_binary(builtins["int2"].oid)
- BinaryInt4Loader.register_binary(builtins["int4"].oid)
- BinaryInt8Loader.register_binary(builtins["int8"].oid)
- BinaryOidLoader.register_binary(builtins["oid"].oid)
- BinaryBoolLoader.register_binary(builtins["bool"].oid)
+ Int2BinaryLoader.register_binary(builtins["int2"].oid)
+ Int4BinaryLoader.register_binary(builtins["int4"].oid)
+ Int8BinaryLoader.register_binary(builtins["int8"].oid)
+ OidBinaryLoader.register_binary(builtins["oid"].oid)
+ BoolBinaryLoader.register_binary(builtins["bool"].oid)
from psycopg3_c cimport libpq
-cdef class StringLoader(CLoader):
+cdef class TextLoader(CLoader):
cdef int is_utf8
cdef object pydecoder
return b
-cdef class TextByteaLoader(CLoader):
+cdef class ByteaLoader(CLoader):
cdef object cload(self, const char *data, size_t length):
cdef size_t len_out
cdef unsigned char *out = libpq.PQunescapeBytea(
return rv
-cdef class BinaryByteaLoader(CLoader):
+cdef class ByteaBinaryLoader(CLoader):
cdef object cload(self, const char *data, size_t length):
return data[:length]
from psycopg3.adapt import Loader
from psycopg3.types import builtins
- StringLoader.register(0) # INVALID_OID
- StringLoader.register(builtins["text"].oid)
- StringLoader.register_binary(builtins["text"].oid)
- StringLoader.register(builtins["varchar"].oid)
- StringLoader.register_binary(builtins["varchar"].oid)
+ TextLoader.register(0) # INVALID_OID
+ TextLoader.register(builtins["text"].oid)
+ TextLoader.register_binary(builtins["text"].oid)
+ TextLoader.register(builtins["varchar"].oid)
+ TextLoader.register_binary(builtins["varchar"].oid)
- TextByteaLoader.register(builtins['bytea'].oid)
- BinaryByteaLoader.register_binary(builtins['bytea'].oid)
+ ByteaLoader.register(builtins['bytea'].oid)
+ ByteaBinaryLoader.register_binary(builtins['bytea'].oid)
import pytest
import psycopg3.types.json
-from psycopg3.types.json import Json, JsonB
+from psycopg3.types.json import Json, Jsonb
from psycopg3.adapt import Format
samples = [
ph = "%s" if fmt_in == Format.TEXT else "%b"
obj = json.loads(val)
cur = conn.cursor()
- cur.execute(f"select {ph} = %s::jsonb", (JsonB(obj), val))
+ cur.execute(f"select {ph} = %s::jsonb", (Jsonb(obj), val))
assert cur.fetchone()[0] is True
@pytest.mark.parametrize("fmt_in", [Format.TEXT, Format.BINARY])
-@pytest.mark.parametrize("wrapper", ["Json", "JsonB"])
+@pytest.mark.parametrize("wrapper", ["Json", "Jsonb"])
def test_json_dump_customise(conn, wrapper, fmt_in):
ph = "%s" if fmt_in == Format.TEXT else "%b"
wrapper = getattr(psycopg3.types.json, wrapper)
@pytest.mark.parametrize("fmt_in", [Format.TEXT, Format.BINARY])
-@pytest.mark.parametrize("wrapper", ["Json", "JsonB"])
+@pytest.mark.parametrize("wrapper", ["Json", "Jsonb"])
def test_json_dump_subclass(conn, wrapper, fmt_in):
ph = "%s" if fmt_in == Format.TEXT else "%b"
wrapper = getattr(psycopg3.types.json, wrapper)
from psycopg3 import sql
from psycopg3.adapt import Transformer, Format
from psycopg3.types import builtins
-from psycopg3.types.numeric import TextFloatLoader
+from psycopg3.types.numeric import FloatLoader
#
)
def test_numeric_as_float(conn, val):
cur = conn.cursor()
- TextFloatLoader.register(builtins["numeric"].oid, cur)
+ FloatLoader.register(builtins["numeric"].oid, cur)
val = Decimal(val)
cur.execute("select %s", (val,))
projects / thirdparty / psycopg.git / commitdiff
