Convert Python object of the type *src* to PostgreSQL representation.
"""
+ format: Format
connection: Optional["BaseConnection"] = None
# A class-wide oid, which will be used by default by instances unless
Convert PostgreSQL objects with OID *oid* to Python objects.
"""
+ format: Format
connection: Optional["BaseConnection"]
def __init__(self, oid: int, context: Optional[AdaptContext] = None):
from typing import Any, Sequence
from .oids import builtins
-from .adapt import Dumper
+from .adapt import Dumper, Format
class DBAPITypeObject:
@Dumper.text(Binary)
class BinaryDumper(Dumper):
+ format = Format.TEXT
oid = builtins["bytea"].oid
def dump(self, obj: Binary) -> bytes:
# they are empty strings, contain curly braces, delimiter characters,
# double quotes, backslashes, or white space, or match the word NULL.
# TODO: recognise only , as delimiter. Should be configured
+
+ format = Format.TEXT
+
_re_needs_quotes = re.compile(
br"""(?xi)
^$ # the empty string
@Dumper.binary(list)
class ListBinaryDumper(BaseListDumper):
+
+ format = Format.BINARY
+
def dump(self, obj: List[Any]) -> bytes:
if not obj:
return _struct_head.pack(0, 0, TEXT_OID)
class ArrayLoader(BaseArrayLoader):
+ format = Format.TEXT
+
# Tokenize an array representation into item and brackets
# TODO: currently recognise only , as delimiter. Should be configured
_re_parse = re.compile(
class ArrayBinaryLoader(BaseArrayLoader):
+
+ format = Format.BINARY
+
def load(self, data: bytes) -> List[Any]:
ndims, hasnull, oid = _struct_head.unpack_from(data[:12])
if not ndims:
class SequenceDumper(Dumper):
+
+ format = Format.TEXT
+
def __init__(self, src: type, context: Optional[AdaptContext] = None):
super().__init__(src, context)
self._tx = Transformer(context)
class BaseCompositeLoader(Loader):
+
+ format = Format.TEXT
+
def __init__(self, oid: int, context: Optional[AdaptContext] = None):
super().__init__(oid, context)
self._tx = Transformer(context)
@Loader.binary(builtins["record"].oid)
-class RecordBinaryLoader(BaseCompositeLoader):
+class RecordBinaryLoader(Loader):
+
+ format = Format.BINARY
_types_set = False
+ def __init__(self, oid: int, context: Optional[AdaptContext] = None):
+ super().__init__(oid, context)
+ self._tx = Transformer(context)
+
def load(self, data: bytes) -> Tuple[Any, ...]:
if not self._types_set:
self._config_types(data)
class CompositeLoader(RecordLoader):
+
+ format = Format.TEXT
factory: Callable[..., Any]
fields_types: List[int]
_types_set = False
class CompositeBinaryLoader(RecordBinaryLoader):
+
+ format = Format.BINARY
factory: Callable[..., Any]
def load(self, data: bytes) -> Any:
from typing import cast, Optional
from ..oids import builtins
-from ..adapt import Dumper, Loader
+from ..adapt import Dumper, Loader, Format
from ..proto import AdaptContext
from ..errors import InterfaceError, DataError
@Dumper.text(date)
class DateDumper(Dumper):
+ format = Format.TEXT
_oid = builtins["date"].oid
def dump(self, obj: date) -> bytes:
@Dumper.text(time)
class TimeDumper(Dumper):
+ format = Format.TEXT
_oid = builtins["timetz"].oid
def dump(self, obj: time) -> bytes:
@Dumper.text(datetime)
class DateTimeDumper(Dumper):
+ format = Format.TEXT
_oid = builtins["timestamptz"].oid
def dump(self, obj: date) -> bytes:
@Dumper.text(timedelta)
class TimeDeltaDumper(Dumper):
+ format = Format.TEXT
_oid = builtins["interval"].oid
def __init__(self, src: type, context: Optional[AdaptContext] = None):
@Loader.text(builtins["date"].oid)
class DateLoader(Loader):
+
+ format = Format.TEXT
+
def __init__(self, oid: int, context: Optional[AdaptContext] = None):
super().__init__(oid, context)
self._format = self._format_from_context()
@Loader.text(builtins["time"].oid)
class TimeLoader(Loader):
+ format = Format.TEXT
_format = "%H:%M:%S.%f"
_format_no_micro = _format.replace(".%f", "")
@Loader.text(builtins["timetz"].oid)
class TimeTzLoader(TimeLoader):
+
+ format = Format.TEXT
_format = "%H:%M:%S.%f%z"
_format_no_micro = _format.replace(".%f", "")
@Loader.text(builtins["timestamp"].oid)
class TimestampLoader(DateLoader):
+
+ format = Format.TEXT
+
def __init__(self, oid: int, context: Optional[AdaptContext] = None):
super().__init__(oid, context)
self._format_no_micro = self._format.replace(".%f", "")
@Loader.text(builtins["timestamptz"].oid)
class TimestamptzLoader(TimestampLoader):
+
+ format = Format.TEXT
+
def __init__(self, oid: int, context: Optional[AdaptContext] = None):
if sys.version_info < (3, 7):
setattr(self, "load", self._load_py36)
@Loader.text(builtins["interval"].oid)
class IntervalLoader(Loader):
+ format = Format.TEXT
+
_re_interval = re.compile(
br"""
(?: (?P<years> [-+]?\d+) \s+ years? \s* )?
from typing import Any, Callable, Optional
from ..oids import builtins
-from ..adapt import Dumper, Loader
+from ..adapt import Dumper, Loader, Format
from ..errors import DataError
JsonDumpsFunction = Callable[[Any], str]
class _JsonDumper(Dumper):
+
+ format = Format.TEXT
+
def dump(self, obj: _JsonWrapper) -> bytes:
return obj.dumps().encode("utf-8")
@Dumper.text(Json)
-@Dumper.binary(Json)
class JsonDumper(_JsonDumper):
+
+ format = Format.TEXT
_oid = builtins["json"].oid
+@Dumper.binary(Json)
+class JsonBinaryDumper(JsonDumper):
+
+ format = Format.BINARY
+
+
@Dumper.text(Jsonb)
class JsonbDumper(_JsonDumper):
+
+ format = Format.TEXT
_oid = builtins["jsonb"].oid
@Dumper.binary(Jsonb)
class JsonbBinaryDumper(JsonbDumper):
+
+ format = Format.BINARY
+
def dump(self, obj: _JsonWrapper) -> bytes:
return b"\x01" + obj.dumps().encode("utf-8")
@Loader.text(builtins["json"].oid)
@Loader.text(builtins["jsonb"].oid)
-@Loader.binary(builtins["json"].oid)
class JsonLoader(Loader):
+
+ format = Format.TEXT
+
def load(self, data: bytes) -> Any:
return json.loads(data)
+@Loader.binary(builtins["json"].oid)
+class JsonBinaryLoader(JsonLoader):
+
+ format = Format.BINARY
+
+
@Loader.binary(builtins["jsonb"].oid)
class JsonbBinaryLoader(Loader):
+
+ format = Format.BINARY
+
def load(self, data: bytes) -> Any:
if data and data[0] != 1:
raise DataError("unknown jsonb binary format: {data[0]}")
from typing import Callable, Optional, Union, TYPE_CHECKING
from ..oids import builtins
-from ..adapt import Dumper, Loader
+from ..adapt import Dumper, Loader, Format
from ..proto import AdaptContext
if TYPE_CHECKING:
@Dumper.text("ipaddress.IPv6Interface")
class InterfaceDumper(Dumper):
+ format = Format.TEXT
_oid = builtins["inet"].oid
def dump(self, obj: Interface) -> bytes:
@Dumper.text("ipaddress.IPv6Network")
class NetworkDumper(Dumper):
+ format = Format.TEXT
_oid = builtins["cidr"].oid
def dump(self, obj: Network) -> bytes:
@Loader.text(builtins["inet"].oid)
class InetLoader(_LazyIpaddress):
+
+ format = Format.TEXT
+
def load(self, data: bytes) -> Union[Address, Interface]:
if b"/" in data:
return ip_interface(data.decode("utf8"))
@Loader.text(builtins["cidr"].oid)
class CidrLoader(_LazyIpaddress):
+
+ format = Format.TEXT
+
def load(self, data: bytes) -> Network:
return ip_network(data.decode("utf8"))
from decimal import Decimal
from ..oids import builtins
-from ..adapt import Dumper, Loader
+from ..adapt import Dumper, Loader, Format
_PackInt = Callable[[int], bytes]
_PackFloat = Callable[[float], bytes]
class NumberDumper(Dumper):
+
+ format = Format.TEXT
+
def dump(self, obj: Any) -> bytes:
return str(obj).encode("utf8")
class SpecialValuesDumper(NumberDumper):
+
_special: Dict[bytes, bytes] = {}
def quote(self, obj: Any) -> bytes:
@Dumper.binary(int)
class IntBinaryDumper(IntDumper):
+
+ format = Format.BINARY
+
def dump(self, obj: int) -> bytes:
return _pack_int8(obj)
@Dumper.text(float)
class FloatDumper(SpecialValuesDumper):
+
+ format = Format.TEXT
_oid = builtins["float8"].oid
_special = {
@Dumper.binary(float)
-class FloatBinaryDumper(NumberDumper):
+class FloatBinaryDumper(Dumper):
+
+ format = Format.BINARY
_oid = builtins["float8"].oid
def dump(self, obj: float) -> bytes:
@Dumper.binary(Int2)
class Int2BinaryDumper(Int2Dumper):
+
+ format = Format.BINARY
+
def dump(self, obj: int) -> bytes:
return _pack_int2(obj)
@Dumper.binary(Int4)
class Int4BinaryDumper(Int4Dumper):
+
+ format = Format.BINARY
+
def dump(self, obj: int) -> bytes:
return _pack_int4(obj)
@Dumper.binary(Int8)
class Int8BinaryDumper(Int8Dumper):
+
+ format = Format.BINARY
+
def dump(self, obj: int) -> bytes:
return _pack_int8(obj)
@Dumper.binary(Oid)
class OidBinaryDumper(OidDumper):
+
+ format = Format.BINARY
+
def dump(self, obj: int) -> bytes:
return _pack_uint4(obj)
@Loader.text(builtins["int8"].oid)
@Loader.text(builtins["oid"].oid)
class IntLoader(Loader):
+
+ format = Format.TEXT
+
def load(self, data: bytes) -> int:
# it supports bytes directly
return int(data)
@Loader.binary(builtins["int2"].oid)
class Int2BinaryLoader(Loader):
+
+ format = Format.BINARY
+
def load(self, data: bytes) -> int:
return _unpack_int2(data)[0]
@Loader.binary(builtins["int4"].oid)
class Int4BinaryLoader(Loader):
+
+ format = Format.BINARY
+
def load(self, data: bytes) -> int:
return _unpack_int4(data)[0]
@Loader.binary(builtins["int8"].oid)
class Int8BinaryLoader(Loader):
+
+ format = Format.BINARY
+
def load(self, data: bytes) -> int:
return _unpack_int8(data)[0]
@Loader.binary(builtins["oid"].oid)
class OidBinaryLoader(Loader):
+
+ format = Format.BINARY
+
def load(self, data: bytes) -> int:
return _unpack_uint4(data)[0]
@Loader.text(builtins["float4"].oid)
@Loader.text(builtins["float8"].oid)
class FloatLoader(Loader):
+
+ format = Format.TEXT
+
def load(self, data: bytes) -> float:
# it supports bytes directly
return float(data)
@Loader.binary(builtins["float4"].oid)
class Float4BinaryLoader(Loader):
+
+ format = Format.BINARY
+
def load(self, data: bytes) -> float:
return _unpack_float4(data)[0]
@Loader.binary(builtins["float8"].oid)
class Float8BinaryLoader(Loader):
+
+ format = Format.BINARY
+
def load(self, data: bytes) -> float:
return _unpack_float8(data)[0]
@Loader.text(builtins["numeric"].oid)
class NumericLoader(Loader):
+
+ format = Format.TEXT
+
def load(self, data: bytes) -> Decimal:
return Decimal(data.decode("utf8"))
# Copyright (C) 2020 The Psycopg Team
from ..oids import builtins
-from ..adapt import Dumper, Loader
+from ..adapt import Dumper, Loader, Format
@Dumper.text(bool)
class BoolDumper(Dumper):
+ format = Format.TEXT
_oid = builtins["bool"].oid
def dump(self, obj: bool) -> bytes:
@Dumper.binary(bool)
class BoolBinaryDumper(Dumper):
+ format = Format.BINARY
_oid = builtins["bool"].oid
def dump(self, obj: bool) -> bytes:
quote(), so it can be used in sql composition.
"""
+ format = Format.TEXT
+
def dump(self, obj: None) -> bytes:
raise NotImplementedError("NULL is passed to Postgres in other ways")
@Loader.text(builtins["bool"].oid)
class BoolLoader(Loader):
+
+ format = Format.TEXT
+
def load(self, data: bytes) -> bool:
return data == b"t"
@Loader.binary(builtins["bool"].oid)
class BoolBinaryLoader(Loader):
+
+ format = Format.BINARY
+
def load(self, data: bytes) -> bool:
return data != b"\x00"
from ..pq import Escaping
from ..oids import builtins, INVALID_OID
-from ..adapt import Dumper, Loader
+from ..adapt import Dumper, Loader, Format
from ..proto import AdaptContext
from ..errors import DataError
@Dumper.binary(str)
class StringBinaryDumper(_StringDumper):
+
+ format = Format.BINARY
+
def dump(self, obj: str) -> bytes:
# the server will raise DataError subclass if the string contains 0x00
return obj.encode(self._encoding)
@Dumper.text(str)
class StringDumper(_StringDumper):
+
+ format = Format.TEXT
+
def dump(self, obj: str) -> bytes:
if "\x00" in obj:
raise DataError(
@Loader.text(builtins["name"].oid)
@Loader.text(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)
class TextLoader(Loader):
+ format = Format.TEXT
_encoding = "utf-8"
def __init__(self, oid: int, context: Optional[AdaptContext] = None):
return data
+@Loader.binary(builtins["bpchar"].oid)
+@Loader.binary(builtins["name"].oid)
+@Loader.binary(builtins["text"].oid)
+@Loader.binary(builtins["varchar"].oid)
+class TextBinaryLoader(TextLoader):
+
+ format = Format.BINARY
+
+
@Dumper.text(bytes)
@Dumper.text(bytearray)
@Dumper.text(memoryview)
class BytesDumper(Dumper):
+ format = Format.TEXT
_oid = builtins["bytea"].oid
def __init__(self, src: type, context: Optional[AdaptContext] = None):
@Dumper.binary(memoryview)
class BytesBinaryDumper(Dumper):
+ format = Format.BINARY
_oid = builtins["bytea"].oid
def dump(
@Loader.text(builtins["bytea"].oid)
class ByteaLoader(Loader):
+
+ format = Format.TEXT
_escaping: "EscapingProto"
def __init__(self, oid: int, context: Optional[AdaptContext] = None):
@Loader.binary(builtins["bytea"].oid)
@Loader.binary(INVALID_OID)
class ByteaBinaryLoader(Loader):
+
+ format = Format.BINARY
+
def load(self, data: bytes) -> bytes:
return data
from typing import Callable, Optional, TYPE_CHECKING
from ..oids import builtins
-from ..adapt import Dumper, Loader
+from ..adapt import Dumper, Loader, Format
from ..proto import AdaptContext
if TYPE_CHECKING:
@Dumper.text("uuid.UUID")
class UUIDDumper(Dumper):
+ format = Format.TEXT
_oid = builtins["uuid"].oid
def dump(self, obj: "uuid.UUID") -> bytes:
@Dumper.binary("uuid.UUID")
class UUIDBinaryDumper(UUIDDumper):
+
+ format = Format.BINARY
+
def dump(self, obj: "uuid.UUID") -> bytes:
return obj.bytes
@Loader.text(builtins["uuid"].oid)
class UUIDLoader(Loader):
+
+ format = Format.TEXT
+
def __init__(self, oid: int, context: Optional[AdaptContext] = None):
super().__init__(oid, context)
global UUID
@Loader.binary(builtins["uuid"].oid)
class UUIDBinaryLoader(UUIDLoader):
+
+ format = Format.BINARY
+
def load(self, data: bytes) -> "uuid.UUID":
return UUID(bytes=data)
def test_dump_connection_ctx(conn):
make_dumper("t").register(str, conn)
- make_dumper("b").register(str, conn, format=Format.BINARY)
+ make_bin_dumper("b").register(str, conn, format=Format.BINARY)
cur = conn.cursor()
cur.execute("select %s, %b", ["hello", "world"])
def test_dump_cursor_ctx(conn):
make_dumper("t").register(str, conn)
- make_dumper("b").register(str, conn, format=Format.BINARY)
+ make_bin_dumper("b").register(str, conn, format=Format.BINARY)
cur = conn.cursor()
make_dumper("tc").register(str, cur)
- make_dumper("bc").register(str, cur, format=Format.BINARY)
+ make_bin_dumper("bc").register(str, cur, format=Format.BINARY)
cur.execute("select %s, %b", ["hello", "world"])
assert cur.fetchone() == ("hellotc", "worldbc")
def test_load_connection_ctx(conn):
make_loader("t").register(TEXT_OID, conn)
- make_loader("b").register(TEXT_OID, conn, format=Format.BINARY)
+ make_bin_loader("b").register(TEXT_OID, conn, format=Format.BINARY)
r = conn.cursor().execute("select 'hello'::text").fetchone()
assert r == ("hellot",)
def test_load_cursor_ctx(conn):
make_loader("t").register(TEXT_OID, conn)
- make_loader("b").register(TEXT_OID, conn, format=Format.BINARY)
+ make_bin_loader("b").register(TEXT_OID, conn, format=Format.BINARY)
cur = conn.cursor()
make_loader("tc").register(TEXT_OID, cur)
- make_loader("bc").register(TEXT_OID, cur, format=Format.BINARY)
+ make_bin_loader("bc").register(TEXT_OID, cur, format=Format.BINARY)
r = cur.execute("select 'hello'::text").fetchone()
assert r == ("hellotc",)
@pytest.mark.parametrize("fmt_out", [Format.TEXT, Format.BINARY])
def test_load_cursor_ctx_nested(conn, sql, obj, fmt_out):
cur = conn.cursor(format=fmt_out)
- make_loader("c").register(TEXT_OID, cur, format=fmt_out)
+ if format == Format.TEXT:
+ make_loader("c").register(TEXT_OID, cur, format=fmt_out)
+ else:
+ make_bin_loader("c").register(TEXT_OID, cur, format=fmt_out)
+
cur.execute(f"select {sql}")
res = cur.fetchone()[0]
assert res == obj
class TestDumper(Dumper):
oid = TEXT_OID
+ format = Format.TEXT
def dump(self, s):
return (s + suffix).encode("ascii")
return TestDumper
+def make_bin_dumper(suffix):
+ cls = make_dumper(suffix)
+ cls.format = Format.BINARY
+ return cls
+
+
def make_loader(suffix):
"""Create a test loader appending a suffix to the data returned."""
class TestLoader(Loader):
+ format = Format.TEXT
+
def load(self, b):
return b.decode("ascii") + suffix
return TestLoader
+
+
+def make_bin_loader(suffix):
+ cls = make_loader(suffix)
+ cls.format = Format.BINARY
+ return cls
projects / thirdparty / psycopg.git / commitdiff
