).format(regtype=cls._to_regtype(conn))
-class SequenceDumper(RecursiveDumper):
+class _SequenceDumper(RecursiveDumper):
def _dump_sequence(
self, obj: Sequence[Any], start: bytes, end: bytes, sep: bytes
) -> bytes:
_re_esc = re.compile(rb"([\\\"])")
-class TupleDumper(SequenceDumper):
+class TupleDumper(_SequenceDumper):
# Should be this, but it doesn't work
# oid = _oids.RECORD_OID
return self._dump_sequence(obj, b"(", b")", b",")
-class TupleBinaryDumper(Dumper):
+class _SequenceBinaryDumper(Dumper):
format = pq.Format.BINARY
# Subclasses must set this info
- _field_types: tuple[int, ...]
+ field_types: tuple[int, ...]
def __init__(self, cls: type, context: abc.AdaptContext | None = None):
super().__init__(cls, context)
# in case the composite contains another composite. Make sure to use
# a separate Transformer instance instead.
self._tx = Transformer(context)
- self._tx.set_dumper_types(self._field_types, self.format)
+ self._tx.set_dumper_types(self.field_types, self.format)
- nfields = len(self._field_types)
+ nfields = len(self.field_types)
self._formats = (PyFormat.from_pq(self.format),) * nfields
def dump(self, obj: tuple[Any, ...]) -> Buffer | None:
adapted = self._tx.dump_sequence(obj, self._formats)
for i in range(len(obj)):
b = adapted[i]
- oid = self._field_types[i]
+ oid = self.field_types[i]
if b is not None:
out += _pack_oidlen(oid, len(b))
out += b
@cache
def _make_binary_dumper(
name: str, oid: int, field_types: tuple[int, ...]
-) -> type[TupleBinaryDumper]:
- return type(
- f"{name.title()}BinaryDumper",
- (TupleBinaryDumper,),
- {"oid": oid, "_field_types": field_types},
- )
+) -> type[_SequenceBinaryDumper]:
+ d = {"oid": oid, "field_types": field_types}
+ return type(f"{name.title()}BinaryDumper", (_SequenceBinaryDumper,), d)
from psycopg.adapt import PyFormat
from psycopg.postgres import types as builtins
from psycopg.types.range import Range
-from psycopg.types.composite import CompositeInfo, TupleBinaryDumper, TupleDumper
-from psycopg.types.composite import register_composite
+from psycopg.types.composite import CompositeInfo, TupleDumper, register_composite
from ..utils import eur
from ..fix_crdb import crdb_skip_message, is_crdb
register_composite(info, conn)
assert issubclass(info.python_type, tuple)
assert info.python_type.__name__ == "testcomp"
- d = conn.adapters.get_dumper(info.python_type, "b")
- assert issubclass(d, TupleBinaryDumper)
- assert d is not TupleBinaryDumper
tc = info.python_type("foo", 42, 3.14)
cur = conn.execute("select pg_typeof(%b)", [tc])