from . import pq
from . import proto
from .pq import Format as Format
-from .oids import TEXT_OID
+from .oids import builtins, TEXT_OID
from .proto import DumpersMap, DumperType, LoadersMap, LoaderType, AdaptContext
if TYPE_CHECKING:
adapters.register_dumper(src, cls)
@classmethod
- def text(cls, src: Union[type, str]) -> Callable[[DumperType], DumperType]:
- def text_(dumper: DumperType) -> DumperType:
- assert dumper.format == Format.TEXT
- dumper.register(src)
- return dumper
-
- return text_
-
- @classmethod
- def binary(
- cls, src: Union[type, str]
+ def builtin(
+ cls, *types: Union[type, str]
) -> Callable[[DumperType], DumperType]:
- def binary_(dumper: DumperType) -> DumperType:
- assert dumper.format == Format.BINARY
- dumper.register(src)
+ """
+ Decorator to mark a dumper class as default for a builtin type.
+ """
+
+ def builtin_(dumper: DumperType) -> DumperType:
+ for src in types:
+ dumper.register(src)
return dumper
- return binary_
+ return builtin_
class Loader(ABC):
adapters.register_loader(oid, cls)
@classmethod
- def text(cls, oid: int) -> Callable[[LoaderType], LoaderType]:
- def text_(loader: LoaderType) -> LoaderType:
- assert loader.format == Format.TEXT
- loader.register(oid)
- return loader
-
- return text_
+ def builtin(
+ cls, *types: Union[int, str]
+ ) -> Callable[[LoaderType], LoaderType]:
+ """
+ Decorator to mark a loader class as default for a builtin type.
+ """
- @classmethod
- def binary(cls, oid: int) -> Callable[[LoaderType], LoaderType]:
- def binary_(loader: LoaderType) -> LoaderType:
- assert loader.format == Format.BINARY
- loader.register(oid)
+ def builtin_(loader: LoaderType) -> LoaderType:
+ for src in types:
+ if isinstance(src, str):
+ src = builtins[src].oid
+ loader.register(src)
return loader
- return binary_
+ return builtin_
class AdaptersMap:
self.obj = obj
-@Dumper.text(Binary)
+@Dumper.builtin(Binary)
class BinaryDumper(Dumper):
format = Format.TEXT
return oid or TEXT_ARRAY_OID
-@Dumper.text(list)
+@Dumper.builtin(list)
class ListDumper(BaseListDumper):
# from https://www.postgresql.org/docs/current/arrays.html#ARRAYS-IO
#
return b"".join(tokens)
-@Dumper.binary(list)
+@Dumper.builtin(list)
class ListBinaryDumper(BaseListDumper):
format = Format.BINARY
from .. import sql
from .. import errors as e
-from ..oids import builtins, TypeInfo, TEXT_OID
+from ..oids import TypeInfo, TEXT_OID
from ..adapt import Format, Dumper, Loader, Transformer
from ..proto import AdaptContext
from . import array
_re_esc = re.compile(br"([\\\"])")
-@Dumper.text(tuple)
+@Dumper.builtin(tuple)
class TupleDumper(SequenceDumper):
# Should be this, but it doesn't work
_re_undouble = re.compile(br'(["\\])\1')
-@Loader.text(builtins["record"].oid)
+@Loader.builtin("record")
class RecordLoader(BaseCompositeLoader):
def load(self, data: bytes) -> Tuple[Any, ...]:
if data == b"()":
_struct_oidlen = struct.Struct("!Ii")
-@Loader.binary(builtins["record"].oid)
+@Loader.builtin("record")
class RecordBinaryLoader(Loader):
format = Format.BINARY
from ..errors import InterfaceError, DataError
-@Dumper.text(date)
+@Dumper.builtin(date)
class DateDumper(Dumper):
format = Format.TEXT
return str(obj).encode("utf8")
-@Dumper.text(time)
+@Dumper.builtin(time)
class TimeDumper(Dumper):
format = Format.TEXT
return str(obj).encode("utf8")
-@Dumper.text(datetime)
+@Dumper.builtin(datetime)
class DateTimeDumper(Dumper):
format = Format.TEXT
return str(obj).encode("utf8")
-@Dumper.text(timedelta)
+@Dumper.builtin(timedelta)
class TimeDeltaDumper(Dumper):
format = Format.TEXT
)
-@Loader.text(builtins["date"].oid)
+@Loader.builtin("date")
class DateLoader(Loader):
format = Format.TEXT
return max(map(len, parts))
-@Loader.text(builtins["time"].oid)
+@Loader.builtin("time")
class TimeLoader(Loader):
format = Format.TEXT
raise exc
-@Loader.text(builtins["timetz"].oid)
+@Loader.builtin("timetz")
class TimeTzLoader(TimeLoader):
format = Format.TEXT
return TimeTzLoader.load(self, data)
-@Loader.text(builtins["timestamp"].oid)
+@Loader.builtin("timestamp")
class TimestampLoader(DateLoader):
format = Format.TEXT
return 0
-@Loader.text(builtins["timestamptz"].oid)
+@Loader.builtin("timestamptz")
class TimestamptzLoader(TimestampLoader):
format = Format.TEXT
)
-@Loader.text(builtins["interval"].oid)
+@Loader.builtin("interval")
class IntervalLoader(Loader):
format = Format.TEXT
return obj.dumps().encode("utf-8")
-@Dumper.text(Json)
+@Dumper.builtin(Json)
class JsonDumper(_JsonDumper):
format = Format.TEXT
_oid = builtins["json"].oid
-@Dumper.binary(Json)
+@Dumper.builtin(Json)
class JsonBinaryDumper(JsonDumper):
format = Format.BINARY
-@Dumper.text(Jsonb)
+@Dumper.builtin(Jsonb)
class JsonbDumper(_JsonDumper):
format = Format.TEXT
_oid = builtins["jsonb"].oid
-@Dumper.binary(Jsonb)
+@Dumper.builtin(Jsonb)
class JsonbBinaryDumper(JsonbDumper):
format = Format.BINARY
return b"\x01" + obj.dumps().encode("utf-8")
-@Loader.text(builtins["json"].oid)
-@Loader.text(builtins["jsonb"].oid)
+@Loader.builtin("json", "jsonb")
class JsonLoader(Loader):
format = Format.TEXT
return json.loads(data)
-@Loader.binary(builtins["json"].oid)
+@Loader.builtin("json")
class JsonBinaryLoader(JsonLoader):
format = Format.BINARY
-@Loader.binary(builtins["jsonb"].oid)
+@Loader.builtin("jsonb")
class JsonbBinaryLoader(Loader):
format = Format.BINARY
ip_network: Callable[[str], Network]
-@Dumper.text("ipaddress.IPv4Address")
-@Dumper.text("ipaddress.IPv6Address")
-@Dumper.text("ipaddress.IPv4Interface")
-@Dumper.text("ipaddress.IPv6Interface")
+@Dumper.builtin(
+ "ipaddress.IPv4Address",
+ "ipaddress.IPv6Address",
+ "ipaddress.IPv4Interface",
+ "ipaddress.IPv6Interface",
+)
class InterfaceDumper(Dumper):
format = Format.TEXT
return str(obj).encode("utf8")
-@Dumper.text("ipaddress.IPv4Network")
-@Dumper.text("ipaddress.IPv6Network")
+@Dumper.builtin("ipaddress.IPv4Network", "ipaddress.IPv6Network")
class NetworkDumper(Dumper):
format = Format.TEXT
imported = True
-@Loader.text(builtins["inet"].oid)
+@Loader.builtin("inet")
class InetLoader(_LazyIpaddress):
format = Format.TEXT
return ip_address(data.decode("utf8"))
-@Loader.text(builtins["cidr"].oid)
+@Loader.builtin("cidr")
class CidrLoader(_LazyIpaddress):
format = Format.TEXT
return value if obj >= 0 else b" " + value
-@Dumper.text(int)
+@Dumper.builtin(int)
class IntDumper(NumberDumper):
_oid = builtins["int8"].oid
-@Dumper.binary(int)
+@Dumper.builtin(int)
class IntBinaryDumper(IntDumper):
format = Format.BINARY
return _pack_int8(obj)
-@Dumper.text(float)
+@Dumper.builtin(float)
class FloatDumper(SpecialValuesDumper):
format = Format.TEXT
}
-@Dumper.binary(float)
+@Dumper.builtin(float)
class FloatBinaryDumper(Dumper):
format = Format.BINARY
return _pack_float8(obj)
-@Dumper.text(Decimal)
+@Dumper.builtin(Decimal)
class DecimalDumper(SpecialValuesDumper):
_oid = builtins["numeric"].oid
}
-@Dumper.text(Int2)
+@Dumper.builtin(Int2)
class Int2Dumper(NumberDumper):
_oid = builtins["int2"].oid
-@Dumper.text(Int4)
+@Dumper.builtin(Int4)
class Int4Dumper(NumberDumper):
_oid = builtins["int4"].oid
-@Dumper.text(Int8)
+@Dumper.builtin(Int8)
class Int8Dumper(NumberDumper):
_oid = builtins["int8"].oid
-@Dumper.text(Oid)
+@Dumper.builtin(Oid)
class OidDumper(NumberDumper):
_oid = builtins["oid"].oid
-@Dumper.binary(Int2)
+@Dumper.builtin(Int2)
class Int2BinaryDumper(Int2Dumper):
format = Format.BINARY
return _pack_int2(obj)
-@Dumper.binary(Int4)
+@Dumper.builtin(Int4)
class Int4BinaryDumper(Int4Dumper):
format = Format.BINARY
return _pack_int4(obj)
-@Dumper.binary(Int8)
+@Dumper.builtin(Int8)
class Int8BinaryDumper(Int8Dumper):
format = Format.BINARY
return _pack_int8(obj)
-@Dumper.binary(Oid)
+@Dumper.builtin(Oid)
class OidBinaryDumper(OidDumper):
format = Format.BINARY
return _pack_uint4(obj)
-@Loader.text(builtins["int2"].oid)
-@Loader.text(builtins["int4"].oid)
-@Loader.text(builtins["int8"].oid)
-@Loader.text(builtins["oid"].oid)
+@Loader.builtin("int2", "int4", "int8", "oid")
class IntLoader(Loader):
format = Format.TEXT
return int(data)
-@Loader.binary(builtins["int2"].oid)
+@Loader.builtin("int2")
class Int2BinaryLoader(Loader):
format = Format.BINARY
return _unpack_int2(data)[0]
-@Loader.binary(builtins["int4"].oid)
+@Loader.builtin("int4")
class Int4BinaryLoader(Loader):
format = Format.BINARY
return _unpack_int4(data)[0]
-@Loader.binary(builtins["int8"].oid)
+@Loader.builtin("int8")
class Int8BinaryLoader(Loader):
format = Format.BINARY
return _unpack_int8(data)[0]
-@Loader.binary(builtins["oid"].oid)
+@Loader.builtin("oid")
class OidBinaryLoader(Loader):
format = Format.BINARY
return _unpack_uint4(data)[0]
-@Loader.text(builtins["float4"].oid)
-@Loader.text(builtins["float8"].oid)
+@Loader.builtin("float4", "float8")
class FloatLoader(Loader):
format = Format.TEXT
return float(data)
-@Loader.binary(builtins["float4"].oid)
+@Loader.builtin("float4")
class Float4BinaryLoader(Loader):
format = Format.BINARY
return _unpack_float4(data)[0]
-@Loader.binary(builtins["float8"].oid)
+@Loader.builtin("float8")
class Float8BinaryLoader(Loader):
format = Format.BINARY
return _unpack_float8(data)[0]
-@Loader.text(builtins["numeric"].oid)
+@Loader.builtin("numeric")
class NumericLoader(Loader):
format = Format.TEXT
# Dumpers for builtin range types
-@Dumper.text(Int4Range)
+@Dumper.builtin(Int4Range)
class Int4RangeDumper(RangeDumper):
_oid = builtins["int4range"].oid
-@Dumper.text(Int8Range)
+@Dumper.builtin(Int8Range)
class Int8RangeDumper(RangeDumper):
_oid = builtins["int8range"].oid
-@Dumper.text(DecimalRange)
+@Dumper.builtin(DecimalRange)
class NumRangeDumper(RangeDumper):
_oid = builtins["numrange"].oid
-@Dumper.text(DateRange)
+@Dumper.builtin(DateRange)
class DateRangeDumper(RangeDumper):
_oid = builtins["daterange"].oid
-@Dumper.text(DateTimeRange)
+@Dumper.builtin(DateTimeRange)
class TimestampRangeDumper(RangeDumper):
_oid = builtins["tsrange"].oid
-@Dumper.text(DateTimeTZRange)
+@Dumper.builtin(DateTimeTZRange)
class TimestampTZRangeDumper(RangeDumper):
_oid = builtins["tstzrange"].oid
# Loaders for builtin range types
-@Loader.text(builtins["int4range"].oid)
+@Loader.builtin("int4range")
class Int4RangeLoader(RangeLoader[int]):
subtype_oid = builtins["int4"].oid
cls = Int4Range
-@Loader.text(builtins["int8range"].oid)
+@Loader.builtin("int8range")
class Int8RangeLoader(RangeLoader[int]):
subtype_oid = builtins["int8"].oid
cls = Int8Range
-@Loader.text(builtins["numrange"].oid)
+@Loader.builtin("numrange")
class NumericRangeLoader(RangeLoader[Decimal]):
subtype_oid = builtins["numeric"].oid
cls = DecimalRange
-@Loader.text(builtins["daterange"].oid)
+@Loader.builtin("daterange")
class DateRangeLoader(RangeLoader[date]):
subtype_oid = builtins["date"].oid
cls = DateRange
-@Loader.text(builtins["tsrange"].oid)
+@Loader.builtin("tsrange")
class TimestampRangeLoader(RangeLoader[datetime]):
subtype_oid = builtins["timestamp"].oid
cls = DateTimeRange
-@Loader.text(builtins["tstzrange"].oid)
+@Loader.builtin("tstzrange")
class TimestampTZRangeLoader(RangeLoader[datetime]):
subtype_oid = builtins["timestamptz"].oid
cls = DateTimeTZRange
from ..adapt import Dumper, Loader, Format
-@Dumper.text(bool)
+@Dumper.builtin(bool)
class BoolDumper(Dumper):
format = Format.TEXT
return b"true" if obj else b"false"
-@Dumper.binary(bool)
+@Dumper.builtin(bool)
class BoolBinaryDumper(Dumper):
format = Format.BINARY
return b"\x01" if obj else b"\x00"
-@Dumper.text(type(None))
+@Dumper.builtin(type(None))
class NoneDumper(Dumper):
"""
Not a complete dumper as it doesn't implement dump(), but it implements
return b"NULL"
-@Loader.text(builtins["bool"].oid)
+@Loader.builtin("bool")
class BoolLoader(Loader):
format = Format.TEXT
return data == b"t"
-@Loader.binary(builtins["bool"].oid)
+@Loader.builtin("bool")
class BoolBinaryLoader(Loader):
format = Format.BINARY
self._encoding = enc
-@Dumper.binary(str)
+@Dumper.builtin(str)
class StringBinaryDumper(_StringDumper):
format = Format.BINARY
return obj.encode(self._encoding)
-@Dumper.text(str)
+@Dumper.builtin(str)
class StringDumper(_StringDumper):
format = Format.TEXT
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.text(builtins["varchar"].oid)
+@Loader.builtin(INVALID_OID, "bpchar", "name", "text", "varchar")
class TextLoader(Loader):
format = Format.TEXT
return data
-@Loader.binary(builtins["bpchar"].oid)
-@Loader.binary(builtins["name"].oid)
-@Loader.binary(builtins["text"].oid)
-@Loader.binary(builtins["varchar"].oid)
+@Loader.builtin("bpchar", "name", "text", "varchar")
class TextBinaryLoader(TextLoader):
format = Format.BINARY
-@Dumper.text(bytes)
-@Dumper.text(bytearray)
-@Dumper.text(memoryview)
+@Dumper.builtin(bytes, bytearray, memoryview)
class BytesDumper(Dumper):
format = Format.TEXT
return self._esc.escape_bytea(obj)
-@Dumper.binary(bytes)
-@Dumper.binary(bytearray)
-@Dumper.binary(memoryview)
+@Dumper.builtin(bytes, bytearray, memoryview)
class BytesBinaryDumper(Dumper):
format = Format.BINARY
return obj
-@Loader.text(builtins["bytea"].oid)
+@Loader.builtin("bytea")
class ByteaLoader(Loader):
format = Format.TEXT
return self._escaping.unescape_bytea(data)
-@Loader.binary(builtins["bytea"].oid)
-@Loader.binary(INVALID_OID)
+@Loader.builtin("bytea", INVALID_OID)
class ByteaBinaryLoader(Loader):
format = Format.BINARY
UUID: Callable[..., "uuid.UUID"]
-@Dumper.text("uuid.UUID")
+@Dumper.builtin("uuid.UUID")
class UUIDDumper(Dumper):
format = Format.TEXT
return obj.hex.encode("utf8")
-@Dumper.binary("uuid.UUID")
+@Dumper.builtin("uuid.UUID")
class UUIDBinaryDumper(UUIDDumper):
format = Format.BINARY
return obj.bytes
-@Loader.text(builtins["uuid"].oid)
+@Loader.builtin("uuid")
class UUIDLoader(Loader):
format = Format.TEXT
return UUID(data.decode("utf8"))
-@Loader.binary(builtins["uuid"].oid)
+@Loader.builtin("uuid")
class UUIDBinaryLoader(UUIDLoader):
format = Format.BINARY
projects / thirdparty / psycopg.git / commitdiff
