Convert Python object of the type *cls* to PostgreSQL representation.
"""
- # A class-wide oid, which will be used by default by instances unless
- # the subclass overrides it in init.
- _oid: int = 0
-
- oid: int
+ oid: int = 0
"""The oid to pass to the server, if known."""
def __init__(self, cls: type, context: Optional[abc.AdaptContext] = None):
context.connection if context else None
)
- self.oid = self._oid
-
def __repr__(self) -> str:
return (
f"<{type(self).__module__}.{type(self).__qualname__}"
NUMBERS_TYPES = (int, float, Decimal)
- _oid = postgres.types["numeric"].array_oid
+ oid = postgres.types["numeric"].array_oid
class ListBinaryDumper(BaseListDumper):
class BoolDumper(Dumper):
format = Format.TEXT
- _oid = postgres.types["bool"].oid
+ oid = postgres.types["bool"].oid
def dump(self, obj: bool) -> bytes:
return b"t" if obj else b"f"
class BoolBinaryDumper(Dumper):
format = Format.BINARY
- _oid = postgres.types["bool"].oid
+ oid = postgres.types["bool"].oid
def dump(self, obj: bool) -> bytes:
return b"\x01" if obj else b"\x00"
class TupleDumper(SequenceDumper):
# Should be this, but it doesn't work
- # _oid = postgres_types["record"].oid
+ # oid = postgres_types["record"].oid
def dump(self, obj: Tuple[Any, ...]) -> bytes:
return self._dump_sequence(obj, b"(", b")", b",")
dumper = type(
f"{info.name.title()}BinaryDumper",
(TupleBinaryDumper,),
- {"_oid": info.oid, "info": info},
+ {"oid": info.oid, "info": info},
)
adapters.register_dumper(factory, dumper)
# Default to the text dumper because it is more flexible
dumper = type(
- f"{info.name.title()}Dumper", (TupleDumper,), {"_oid": info.oid}
+ f"{info.name.title()}Dumper", (TupleDumper,), {"oid": info.oid}
)
adapters.register_dumper(factory, dumper)
class DateDumper(Dumper):
format = Format.TEXT
- _oid = postgres.types["date"].oid
+ oid = postgres.types["date"].oid
def dump(self, obj: date) -> bytes:
# NOTE: whatever the PostgreSQL DateStyle input format (DMY, MDY, YMD)
class DateBinaryDumper(Dumper):
format = Format.BINARY
- _oid = postgres.types["date"].oid
+ oid = postgres.types["date"].oid
def dump(self, obj: date) -> bytes:
days = obj.toordinal() - _pg_date_epoch_days
class TimeDumper(_BaseTimeTextDumper):
- _oid = postgres.types["time"].oid
+ oid = postgres.types["time"].oid
def upgrade(self, obj: time, format: PyFormat) -> Dumper:
if not obj.tzinfo:
class TimeTzDumper(_BaseTimeTextDumper):
- _oid = postgres.types["timetz"].oid
+ oid = postgres.types["timetz"].oid
class TimeBinaryDumper(_BaseTimeDumper):
format = Format.BINARY
- _oid = postgres.types["time"].oid
+ oid = postgres.types["time"].oid
def dump(self, obj: time) -> bytes:
us = obj.microsecond + 1_000_000 * (
class TimeTzBinaryDumper(_BaseTimeDumper):
format = Format.BINARY
- _oid = postgres.types["timetz"].oid
+ oid = postgres.types["timetz"].oid
def dump(self, obj: time) -> bytes:
us = obj.microsecond + 1_000_000 * (
class DatetimeDumper(_BaseDatetimeTextDumper):
- _oid = postgres.types["timestamptz"].oid
+ oid = postgres.types["timestamptz"].oid
def upgrade(self, obj: datetime, format: PyFormat) -> Dumper:
if obj.tzinfo:
class DatetimeNoTzDumper(_BaseDatetimeTextDumper):
- _oid = postgres.types["timestamp"].oid
+ oid = postgres.types["timestamp"].oid
class DatetimeBinaryDumper(_BaseDatetimeDumper):
format = Format.BINARY
- _oid = postgres.types["timestamptz"].oid
+ oid = postgres.types["timestamptz"].oid
def dump(self, obj: datetime) -> bytes:
delta = obj - _pg_datetimetz_epoch
class DatetimeNoTzBinaryDumper(_BaseDatetimeDumper):
format = Format.BINARY
- _oid = postgres.types["timestamp"].oid
+ oid = postgres.types["timestamp"].oid
def dump(self, obj: datetime) -> bytes:
delta = obj - _pg_datetime_epoch
class TimedeltaDumper(Dumper):
format = Format.TEXT
- _oid = postgres.types["interval"].oid
+ oid = postgres.types["interval"].oid
def __init__(self, cls: type, context: Optional[AdaptContext] = None):
super().__init__(cls, context)
class TimedeltaBinaryDumper(Dumper):
format = Format.BINARY
- _oid = postgres.types["interval"].oid
+ oid = postgres.types["interval"].oid
def dump(self, obj: timedelta) -> bytes:
micros = 1_000_000 * obj.seconds + obj.microseconds
# Generate and register a customized text dumper
dumper: Type[BaseHstoreDumper] = type(
- "HstoreDumper", (BaseHstoreDumper,), {"_oid": info.oid}
+ "HstoreDumper", (BaseHstoreDumper,), {"oid": info.oid}
)
adapters.register_dumper(dict, dumper)
class JsonDumper(_JsonDumper):
format = Format.TEXT
- _oid = postgres.types["json"].oid
+ oid = postgres.types["json"].oid
class JsonBinaryDumper(_JsonDumper):
format = Format.BINARY
- _oid = postgres.types["json"].oid
+ oid = postgres.types["json"].oid
class JsonbDumper(_JsonDumper):
format = Format.TEXT
- _oid = postgres.types["jsonb"].oid
+ oid = postgres.types["jsonb"].oid
class JsonbBinaryDumper(_JsonDumper):
format = Format.BINARY
- _oid = postgres.types["jsonb"].oid
+ oid = postgres.types["jsonb"].oid
def dump(self, obj: _JsonWrapper) -> bytes:
dumps = obj.dumps or self.dumps
class InterfaceDumper(Dumper):
format = Format.TEXT
- _oid = postgres.types["inet"].oid
+ oid = postgres.types["inet"].oid
def dump(self, obj: Interface) -> bytes:
return str(obj).encode("utf8")
class NetworkDumper(Dumper):
format = Format.TEXT
- _oid = postgres.types["cidr"].oid
+ oid = postgres.types["cidr"].oid
def dump(self, obj: Network) -> bytes:
return str(obj).encode("utf8")
class AddressBinaryDumper(Dumper):
format = Format.BINARY
- _oid = postgres.types["inet"].oid
+ oid = postgres.types["inet"].oid
def dump(self, obj: Address) -> bytes:
packed = obj.packed
class InterfaceBinaryDumper(Dumper):
format = Format.BINARY
- _oid = postgres.types["inet"].oid
+ oid = postgres.types["inet"].oid
def dump(self, obj: Interface) -> bytes:
packed = obj.packed
class NetworkBinaryDumper(Dumper):
format = Format.BINARY
- _oid = postgres.types["cidr"].oid
+ oid = postgres.types["cidr"].oid
def dump(self, obj: Network) -> bytes:
packed = obj.network_address.packed
class FloatDumper(_SpecialValuesDumper):
format = Format.TEXT
- _oid = postgres.types["float8"].oid
+ oid = postgres.types["float8"].oid
_special = {
b"inf": b"'Infinity'::float8",
class Float4Dumper(FloatDumper):
- _oid = postgres.types["float4"].oid
+ oid = postgres.types["float4"].oid
class FloatBinaryDumper(Dumper):
format = Format.BINARY
- _oid = postgres.types["float8"].oid
+ oid = postgres.types["float8"].oid
def dump(self, obj: float) -> bytes:
return pack_float8(obj)
class Float4BinaryDumper(FloatBinaryDumper):
- _oid = postgres.types["float4"].oid
+ oid = postgres.types["float4"].oid
def dump(self, obj: float) -> bytes:
return pack_float4(obj)
class DecimalDumper(_SpecialValuesDumper):
- _oid = postgres.types["numeric"].oid
+ oid = postgres.types["numeric"].oid
def dump(self, obj: Decimal) -> bytes:
if obj.is_nan():
class Int2Dumper(_NumberDumper):
- _oid = postgres.types["int2"].oid
+ oid = postgres.types["int2"].oid
class Int4Dumper(_NumberDumper):
- _oid = postgres.types["int4"].oid
+ oid = postgres.types["int4"].oid
class Int8Dumper(_NumberDumper):
- _oid = postgres.types["int8"].oid
+ oid = postgres.types["int8"].oid
class IntNumericDumper(_NumberDumper):
- _oid = postgres.types["numeric"].oid
+ oid = postgres.types["numeric"].oid
class OidDumper(_NumberDumper):
- _oid = postgres.types["oid"].oid
+ oid = postgres.types["oid"].oid
class IntDumper(Dumper):
class DecimalBinaryDumper(Dumper):
format = Format.BINARY
- _oid = postgres.types["numeric"].oid
+ oid = postgres.types["numeric"].oid
def dump(self, obj: Decimal) -> Union[bytearray, bytes]:
sign, digits, exp = obj.as_tuple()
class Int4RangeDumper(RangeDumper):
- _oid = postgres.types["int4range"].oid
+ oid = postgres.types["int4range"].oid
class Int8RangeDumper(RangeDumper):
- _oid = postgres.types["int8range"].oid
+ oid = postgres.types["int8range"].oid
class NumericRangeDumper(RangeDumper):
- _oid = postgres.types["numrange"].oid
+ oid = postgres.types["numrange"].oid
class DateRangeDumper(RangeDumper):
- _oid = postgres.types["daterange"].oid
+ oid = postgres.types["daterange"].oid
class TimestampRangeDumper(RangeDumper):
- _oid = postgres.types["tsrange"].oid
+ oid = postgres.types["tsrange"].oid
class TimestamptzRangeDumper(RangeDumper):
- _oid = postgres.types["tstzrange"].oid
+ oid = postgres.types["tstzrange"].oid
# Binary dumpers for builtin range types wrappers
class Int4RangeBinaryDumper(RangeBinaryDumper):
- _oid = postgres.types["int4range"].oid
+ oid = postgres.types["int4range"].oid
class Int8RangeBinaryDumper(RangeBinaryDumper):
- _oid = postgres.types["int8range"].oid
+ oid = postgres.types["int8range"].oid
class NumericRangeBinaryDumper(RangeBinaryDumper):
- _oid = postgres.types["numrange"].oid
+ oid = postgres.types["numrange"].oid
class DateRangeBinaryDumper(RangeBinaryDumper):
- _oid = postgres.types["daterange"].oid
+ oid = postgres.types["daterange"].oid
class TimestampRangeBinaryDumper(RangeBinaryDumper):
- _oid = postgres.types["tsrange"].oid
+ oid = postgres.types["tsrange"].oid
class TimestamptzRangeBinaryDumper(RangeBinaryDumper):
- _oid = postgres.types["tstzrange"].oid
+ oid = postgres.types["tstzrange"].oid
# Text loaders for builtin range types
class StrBinaryDumper(_BaseStrDumper):
format = Format.BINARY
- _oid = postgres.types["text"].oid
+ oid = postgres.types["text"].oid
def dump(self, obj: str) -> bytes:
# the server will raise DataError subclass if the string contains 0x00
text oid is required, for instance with variadic functions.
"""
- _oid = postgres.types["text"].oid
+ oid = postgres.types["text"].oid
class StrDumperUnknown(_StrDumper):
class BytesDumper(Dumper):
format = Format.TEXT
- _oid = postgres.types["bytea"].oid
+ oid = postgres.types["bytea"].oid
_qprefix = b""
def __init__(self, cls: type, context: Optional[AdaptContext] = None):
class BytesBinaryDumper(Dumper):
format = Format.BINARY
- _oid = postgres.types["bytea"].oid
+ oid = postgres.types["bytea"].oid
def dump(self, obj: Buffer) -> Buffer:
return obj
class UUIDDumper(Dumper):
format = Format.TEXT
- _oid = postgres.types["uuid"].oid
+ oid = postgres.types["uuid"].oid
def dump(self, obj: "uuid.UUID") -> bytes:
return obj.hex.encode("utf8")
class MyStrDumper:
format = pq.Format.TEXT
+ oid = 25 # text
def __init__(self, cls: type, context: Optional[AdaptContext] = None):
self._cls = cls
- self.oid = 25 # text
def dump(self, obj: str) -> bytes:
return (obj * 2).encode("utf-8")
class BoxDumper(Dumper):
format = pq.Format.TEXT
- _oid = psycopg.postgres.types["box"].oid
+ oid = psycopg.postgres.types["box"].oid
def dump(self, box):
return ("(%s,%s),(%s,%s)" % box.coords).encode("utf8")
projects / thirdparty / psycopg.git / commitdiff
