[flake8]
per-file-ignores =
psycopg3/__init__.py: F401
+ psycopg3/types/__init__.py: F401
# Copyright (C) 2020 The Psycopg Team
from . import pq
+from . import types
from .copy import Copy, AsyncCopy
+from .adapt import global_adapters
from .cursor import AsyncCursor, Cursor
from .errors import Warning, Error, InterfaceError, DatabaseError
from .errors import DataError, OperationalError, IntegrityError
from .connection import AsyncConnection, Connection, Notify
from .transaction import Rollback, Transaction, AsyncTransaction
-from .dbapi20 import BINARY, DATETIME, NUMBER, ROWID, STRING
+from .dbapi20 import BINARY, DATETIME, NUMBER, ROWID, STRING, BinaryDumper
from .dbapi20 import Binary, Date, DateFromTicks, Time, TimeFromTicks
from .dbapi20 import Timestamp, TimestampFromTicks
from .version import __version__
# register default adapters
-from . import types
+types.register_default_globals(global_adapters)
# DBAPI compliancy
connect = Connection.connect
apilevel = "2.0"
threadsafety = 2
paramstyle = "pyformat"
+BinaryDumper.register(Binary, global_adapters) # dbapi20
# Override adapters with fast version if available
# Copyright (C) 2020 The Psycopg Team
from abc import ABC, abstractmethod
-from typing import Any, Dict, Callable, List, Optional, Type, Union
+from typing import Any, Dict, List, Optional, Type, Union
from typing import TYPE_CHECKING
from . import pq
from . import proto
adapters = context.adapters if context else global_adapters
adapters.register_dumper(cls, this_cls)
- @classmethod
- def builtin(
- cls, *types: Union[type, str]
- ) -> Callable[[Type["Dumper"]], Type["Dumper"]]:
- """
- Decorator to mark a dumper class as default for a builtin type.
- """
-
- def builtin_(dumper: Type["Dumper"]) -> Type["Dumper"]:
- for cls in types:
- dumper.register(cls)
- return dumper
-
- return builtin_
-
class Loader(ABC):
"""
@classmethod
def register(
- cls, oid: int, context: Optional[AdaptContext] = None
+ cls, oid: Union[int, str], context: Optional[AdaptContext] = None
) -> None:
"""
Configure *context* to use this loader to convert values with OID *oid*.
"""
+ if isinstance(oid, str):
+ oid = builtins[oid].oid
adapters = context.adapters if context else global_adapters
adapters.register_loader(oid, cls)
- @classmethod
- def builtin(
- cls, *types: Union[int, str]
- ) -> Callable[[Type["Loader"]], Type["Loader"]]:
- """
- Decorator to mark a loader class as default for a builtin type.
- """
-
- def builtin_(loader: Type["Loader"]) -> Type["Loader"]:
- for cls in types:
- if isinstance(cls, str):
- cls = builtins[cls].oid
- loader.register(cls)
- return loader
-
- return builtin_
-
-class AdaptersMap:
+class AdaptersMap(AdaptContext):
"""
Map oids to Loaders and types to Dumpers.
self._loaders = [{}, {}]
self._own_loaders = [True, True]
+ # implement the AdaptContext protocol too
+ @property
+ def adapters(self) -> "AdaptersMap":
+ return self
+
+ @property
+ def connection(self) -> Optional["BaseConnection"]:
+ return None
+
def register_dumper(
self, cls: Union[type, str], dumper: Type[Dumper]
) -> None:
if scls in dumpers:
return dumpers[scls]
- # If the adapter is not found, look for its name as a string
- for scls in cls.__mro__:
+ # If the adapter is not found, look for its name as a string
fqn = scls.__module__ + "." + scls.__qualname__
if fqn in dumpers:
# Replace the class name with the class itself
self.obj = obj
-@Dumper.builtin(Binary)
class BinaryDumper(Dumper):
format = Format.TEXT
# Copyright (C) 2020 The Psycopg Team
-
-from ..oids import builtins
+from ..oids import builtins, INVALID_OID
+from ..proto import AdaptContext
# Register default adapters
-from . import array, composite, date, json, network, numeric # noqa
-from . import range, singletons, text, uuid # noqa
+from . import array, composite
+from . import range
+
+# Wrapper objects
+from .numeric import Int2, Int4, Int8, Oid
+from .json import Json, Jsonb
+from .range import Range, Int4Range, Int8Range, DecimalRange
+from .range import DateRange, DateTimeRange, DateTimeTZRange
+
+# Supper objects
+from .range import RangeInfo
+from .composite import CompositeInfo
+
+# Adapter objects
+from .text import (
+ StringDumper,
+ StringBinaryDumper,
+ TextLoader,
+ TextBinaryLoader,
+ BytesDumper,
+ BytesBinaryDumper,
+ ByteaLoader,
+ ByteaBinaryLoader,
+)
+from .numeric import (
+ IntDumper,
+ FloatDumper,
+ FloatBinaryDumper,
+ DecimalDumper,
+ Int2Dumper,
+ Int4Dumper,
+ Int8Dumper,
+ OidDumper,
+ Int2BinaryDumper,
+ Int4BinaryDumper,
+ Int8BinaryDumper,
+ OidBinaryDumper,
+ IntLoader,
+ Int2BinaryLoader,
+ Int4BinaryLoader,
+ Int8BinaryLoader,
+ OidBinaryLoader,
+ FloatLoader,
+ Float4BinaryLoader,
+ Float8BinaryLoader,
+ NumericLoader,
+)
+from .singletons import (
+ BoolDumper,
+ BoolBinaryDumper,
+ NoneDumper,
+ BoolLoader,
+ BoolBinaryLoader,
+)
+from .date import (
+ DateDumper,
+ TimeDumper,
+ DateTimeDumper,
+ TimeDeltaDumper,
+ DateLoader,
+ TimeLoader,
+ TimeTzLoader,
+ TimestampLoader,
+ TimestamptzLoader,
+ IntervalLoader,
+)
+from .json import (
+ JsonDumper,
+ JsonBinaryDumper,
+ JsonbDumper,
+ JsonbBinaryDumper,
+ JsonLoader,
+ JsonBinaryLoader,
+ JsonbBinaryLoader,
+)
+from .uuid import (
+ UUIDDumper,
+ UUIDBinaryDumper,
+ UUIDLoader,
+ UUIDBinaryLoader,
+)
+from .network import (
+ InterfaceDumper,
+ NetworkDumper,
+ InetLoader,
+ CidrLoader,
+)
+from .range import (
+ RangeDumper,
+ Int4RangeDumper,
+ Int8RangeDumper,
+ NumRangeDumper,
+ DateRangeDumper,
+ TimestampRangeDumper,
+ TimestampTZRangeDumper,
+ RangeLoader,
+ Int4RangeLoader,
+ Int8RangeLoader,
+ NumericRangeLoader,
+ DateRangeLoader,
+ TimestampRangeLoader,
+ TimestampTZRangeLoader,
+)
+from .array import (
+ ListDumper,
+ ListBinaryDumper,
+)
+from .composite import (
+ TupleDumper,
+ RecordLoader,
+ RecordBinaryLoader,
+ CompositeLoader,
+ CompositeBinaryLoader,
+)
+
+
+def register_default_globals(ctx: AdaptContext) -> None:
+ StringDumper.register(str, ctx)
+ StringBinaryDumper.register(str, ctx)
+ TextLoader.register(INVALID_OID, ctx)
+ TextLoader.register("bpchar", ctx)
+ TextLoader.register("name", ctx)
+ TextLoader.register("text", ctx)
+ TextLoader.register("varchar", ctx)
+ TextBinaryLoader.register("bpchar", ctx)
+ TextBinaryLoader.register("name", ctx)
+ TextBinaryLoader.register("text", ctx)
+ TextBinaryLoader.register("varchar", ctx)
+
+ BytesDumper.register(bytes, ctx)
+ BytesDumper.register(bytearray, ctx)
+ BytesDumper.register(memoryview, ctx)
+ BytesBinaryDumper.register(bytes, ctx)
+ BytesBinaryDumper.register(bytearray, ctx)
+ BytesBinaryDumper.register(memoryview, ctx)
+ ByteaLoader.register("bytea", ctx)
+ ByteaBinaryLoader.register(INVALID_OID, ctx)
+ ByteaBinaryLoader.register("bytea", ctx)
+
+ IntDumper.register(int, ctx)
+ FloatDumper.register(float, ctx)
+ Int8BinaryDumper.register(int, ctx)
+ FloatBinaryDumper.register(float, ctx)
+ DecimalDumper.register("decimal.Decimal", ctx)
+ Int2Dumper.register(Int2, ctx)
+ Int4Dumper.register(Int4, ctx)
+ Int8Dumper.register(Int8, ctx)
+ OidDumper.register(Oid, ctx)
+ Int2BinaryDumper.register(Int2, ctx)
+ Int4BinaryDumper.register(Int4, ctx)
+ Int8BinaryDumper.register(Int8, ctx)
+ OidBinaryDumper.register(Oid, ctx)
+ IntLoader.register("int2", ctx)
+ IntLoader.register("int4", ctx)
+ IntLoader.register("int8", ctx)
+ IntLoader.register("oid", ctx)
+ Int2BinaryLoader.register("int2", ctx)
+ Int4BinaryLoader.register("int4", ctx)
+ Int8BinaryLoader.register("int8", ctx)
+ OidBinaryLoader.register("oid", ctx)
+ FloatLoader.register("float4", ctx)
+ FloatLoader.register("float8", ctx)
+ Float4BinaryLoader.register("float4", ctx)
+ Float8BinaryLoader.register("float8", ctx)
+ NumericLoader.register("numeric", ctx)
+
+ BoolDumper.register(bool, ctx)
+ BoolBinaryDumper.register(bool, ctx)
+ NoneDumper.register(type(None), ctx)
+ BoolLoader.register("bool", ctx)
+ BoolBinaryLoader.register("bool", ctx)
+
+ DateDumper.register("datetime.date", ctx)
+ TimeDumper.register("datetime.time", ctx)
+ DateTimeDumper.register("datetime.datetime", ctx)
+ TimeDeltaDumper.register("datetime.timedelta", ctx)
+ DateLoader.register("date", ctx)
+ TimeLoader.register("time", ctx)
+ TimeTzLoader.register("timetz", ctx)
+ TimestampLoader.register("timestamp", ctx)
+ TimestamptzLoader.register("timestamptz", ctx)
+ IntervalLoader.register("interval", ctx)
+
+ JsonDumper.register(Json, ctx)
+ JsonBinaryDumper.register(Json, ctx)
+ JsonbDumper.register(Jsonb, ctx)
+ JsonbBinaryDumper.register(Jsonb, ctx)
+ JsonLoader.register("json", ctx)
+ JsonLoader.register("jsonb", ctx)
+ JsonBinaryLoader.register("json", ctx)
+ JsonbBinaryLoader.register("jsonb", ctx)
+
+ UUIDDumper.register("uuid.UUID", ctx)
+ UUIDBinaryDumper.register("uuid.UUID", ctx)
+ UUIDLoader.register("uuid", ctx)
+ UUIDBinaryLoader.register("uuid", ctx)
+
+ InterfaceDumper.register("ipaddress.IPv4Address", ctx)
+ InterfaceDumper.register("ipaddress.IPv6Address", ctx)
+ InterfaceDumper.register("ipaddress.IPv4Interface", ctx)
+ InterfaceDumper.register("ipaddress.IPv6Interface", ctx)
+ NetworkDumper.register("ipaddress.IPv4Network", ctx)
+ NetworkDumper.register("ipaddress.IPv6Network", ctx)
+ InetLoader.register("inet", ctx)
+ CidrLoader.register("cidr", ctx)
+
+ Int4RangeDumper.register(Int4Range, ctx)
+ Int8RangeDumper.register(Int8Range, ctx)
+ NumRangeDumper.register(DecimalRange, ctx)
+ DateRangeDumper.register(DateRange, ctx)
+ TimestampRangeDumper.register(DateTimeRange, ctx)
+ TimestampTZRangeDumper.register(DateTimeTZRange, ctx)
+ Int4RangeLoader.register("int4range", ctx)
+ Int8RangeLoader.register("int8range", ctx)
+ NumericRangeLoader.register("numrange", ctx)
+ DateRangeLoader.register("daterange", ctx)
+ TimestampRangeLoader.register("tsrange", ctx)
+ TimestampTZRangeLoader.register("tstzrange", ctx)
-# Register associations with array oids
-array.register_all_arrays()
+ ListDumper.register(list, ctx)
+ ListBinaryDumper.register(list, ctx)
+ TupleDumper.register(tuple, ctx)
+ RecordLoader.register("record", ctx)
+ RecordBinaryLoader.register("record", ctx)
-__all__ = ["builtins"]
+ array.register_all_arrays()
return oid or TEXT_ARRAY_OID
-@Dumper.builtin(list)
class ListDumper(BaseListDumper):
# from https://www.postgresql.org/docs/current/arrays.html#ARRAYS-IO
#
return b"".join(tokens)
-@Dumper.builtin(list)
class ListBinaryDumper(BaseListDumper):
format = Format.BINARY
_re_esc = re.compile(br"([\\\"])")
-@Dumper.builtin(tuple)
class TupleDumper(SequenceDumper):
# Should be this, but it doesn't work
_re_undouble = re.compile(br'(["\\])\1')
-@Loader.builtin("record")
class RecordLoader(BaseCompositeLoader):
def load(self, data: bytes) -> Tuple[Any, ...]:
if data == b"()":
_struct_oidlen = struct.Struct("!Ii")
-@Loader.builtin("record")
class RecordBinaryLoader(Loader):
format = Format.BINARY
from ..errors import InterfaceError, DataError
-@Dumper.builtin(date)
class DateDumper(Dumper):
format = Format.TEXT
return str(obj).encode("utf8")
-@Dumper.builtin(time)
class TimeDumper(Dumper):
format = Format.TEXT
return str(obj).encode("utf8")
-@Dumper.builtin(datetime)
class DateTimeDumper(Dumper):
format = Format.TEXT
return str(obj).encode("utf8")
-@Dumper.builtin(timedelta)
class TimeDeltaDumper(Dumper):
format = Format.TEXT
)
-@Loader.builtin("date")
class DateLoader(Loader):
format = Format.TEXT
return max(map(len, parts))
-@Loader.builtin("time")
class TimeLoader(Loader):
format = Format.TEXT
raise exc
-@Loader.builtin("timetz")
class TimeTzLoader(TimeLoader):
format = Format.TEXT
return TimeTzLoader.load(self, data)
-@Loader.builtin("timestamp")
class TimestampLoader(DateLoader):
format = Format.TEXT
return 0
-@Loader.builtin("timestamptz")
class TimestamptzLoader(TimestampLoader):
format = Format.TEXT
)
-@Loader.builtin("interval")
class IntervalLoader(Loader):
format = Format.TEXT
return obj.dumps().encode("utf-8")
-@Dumper.builtin(Json)
class JsonDumper(_JsonDumper):
format = Format.TEXT
_oid = builtins["json"].oid
-@Dumper.builtin(Json)
class JsonBinaryDumper(JsonDumper):
format = Format.BINARY
-@Dumper.builtin(Jsonb)
class JsonbDumper(_JsonDumper):
format = Format.TEXT
_oid = builtins["jsonb"].oid
-@Dumper.builtin(Jsonb)
class JsonbBinaryDumper(JsonbDumper):
format = Format.BINARY
return b"\x01" + obj.dumps().encode("utf-8")
-@Loader.builtin("json", "jsonb")
class JsonLoader(Loader):
format = Format.TEXT
return json.loads(data)
-@Loader.builtin("json")
class JsonBinaryLoader(JsonLoader):
format = Format.BINARY
-@Loader.builtin("jsonb")
class JsonbBinaryLoader(Loader):
format = Format.BINARY
ip_network: Callable[[str], Network]
-@Dumper.builtin(
- "ipaddress.IPv4Address",
- "ipaddress.IPv6Address",
- "ipaddress.IPv4Interface",
- "ipaddress.IPv6Interface",
-)
class InterfaceDumper(Dumper):
format = Format.TEXT
return str(obj).encode("utf8")
-@Dumper.builtin("ipaddress.IPv4Network", "ipaddress.IPv6Network")
class NetworkDumper(Dumper):
format = Format.TEXT
imported = True
-@Loader.builtin("inet")
class InetLoader(_LazyIpaddress):
format = Format.TEXT
return ip_address(data.decode("utf8"))
-@Loader.builtin("cidr")
class CidrLoader(_LazyIpaddress):
format = Format.TEXT
return value if obj >= 0 else b" " + value
-@Dumper.builtin(int)
class IntDumper(NumberDumper):
_oid = builtins["int8"].oid
-@Dumper.builtin(float)
class FloatDumper(SpecialValuesDumper):
format = Format.TEXT
}
-@Dumper.builtin(float)
class FloatBinaryDumper(Dumper):
format = Format.BINARY
return _pack_float8(obj)
-@Dumper.builtin(Decimal)
class DecimalDumper(SpecialValuesDumper):
_oid = builtins["numeric"].oid
}
-@Dumper.builtin(Int2)
class Int2Dumper(NumberDumper):
_oid = builtins["int2"].oid
-@Dumper.builtin(Int4)
class Int4Dumper(NumberDumper):
_oid = builtins["int4"].oid
-@Dumper.builtin(Int8)
class Int8Dumper(NumberDumper):
_oid = builtins["int8"].oid
-@Dumper.builtin(Oid)
class OidDumper(NumberDumper):
_oid = builtins["oid"].oid
-@Dumper.builtin(Int2)
class Int2BinaryDumper(Int2Dumper):
format = Format.BINARY
return _pack_int2(obj)
-@Dumper.builtin(Int4)
class Int4BinaryDumper(Int4Dumper):
format = Format.BINARY
return _pack_int4(obj)
-@Dumper.builtin(int, Int8)
class Int8BinaryDumper(Int8Dumper):
format = Format.BINARY
return _pack_int8(obj)
-@Dumper.builtin(Oid)
class OidBinaryDumper(OidDumper):
format = Format.BINARY
return _pack_uint4(obj)
-@Loader.builtin("int2", "int4", "int8", "oid")
class IntLoader(Loader):
format = Format.TEXT
return int(data)
-@Loader.builtin("int2")
class Int2BinaryLoader(Loader):
format = Format.BINARY
return _unpack_int2(data)[0]
-@Loader.builtin("int4")
class Int4BinaryLoader(Loader):
format = Format.BINARY
return _unpack_int4(data)[0]
-@Loader.builtin("int8")
class Int8BinaryLoader(Loader):
format = Format.BINARY
return _unpack_int8(data)[0]
-@Loader.builtin("oid")
class OidBinaryLoader(Loader):
format = Format.BINARY
return _unpack_uint4(data)[0]
-@Loader.builtin("float4", "float8")
class FloatLoader(Loader):
format = Format.TEXT
return float(data)
-@Loader.builtin("float4")
class Float4BinaryLoader(Loader):
format = Format.BINARY
return _unpack_float4(data)[0]
-@Loader.builtin("float8")
class Float8BinaryLoader(Loader):
format = Format.BINARY
return _unpack_float8(data)[0]
-@Loader.builtin("numeric")
class NumericLoader(Loader):
format = Format.TEXT
# Dumpers for builtin range types
-@Dumper.builtin(Int4Range)
class Int4RangeDumper(RangeDumper):
_oid = builtins["int4range"].oid
-@Dumper.builtin(Int8Range)
class Int8RangeDumper(RangeDumper):
_oid = builtins["int8range"].oid
-@Dumper.builtin(DecimalRange)
class NumRangeDumper(RangeDumper):
_oid = builtins["numrange"].oid
-@Dumper.builtin(DateRange)
class DateRangeDumper(RangeDumper):
_oid = builtins["daterange"].oid
-@Dumper.builtin(DateTimeRange)
class TimestampRangeDumper(RangeDumper):
_oid = builtins["tsrange"].oid
-@Dumper.builtin(DateTimeTZRange)
class TimestampTZRangeDumper(RangeDumper):
_oid = builtins["tstzrange"].oid
# Loaders for builtin range types
-@Loader.builtin("int4range")
class Int4RangeLoader(RangeLoader[int]):
subtype_oid = builtins["int4"].oid
cls = Int4Range
-@Loader.builtin("int8range")
class Int8RangeLoader(RangeLoader[int]):
subtype_oid = builtins["int8"].oid
cls = Int8Range
-@Loader.builtin("numrange")
class NumericRangeLoader(RangeLoader[Decimal]):
subtype_oid = builtins["numeric"].oid
cls = DecimalRange
-@Loader.builtin("daterange")
class DateRangeLoader(RangeLoader[date]):
subtype_oid = builtins["date"].oid
cls = DateRange
-@Loader.builtin("tsrange")
class TimestampRangeLoader(RangeLoader[datetime]):
subtype_oid = builtins["timestamp"].oid
cls = DateTimeRange
-@Loader.builtin("tstzrange")
class TimestampTZRangeLoader(RangeLoader[datetime]):
subtype_oid = builtins["timestamptz"].oid
cls = DateTimeTZRange
from ..adapt import Dumper, Loader, Format
-@Dumper.builtin(bool)
class BoolDumper(Dumper):
format = Format.TEXT
return b"true" if obj else b"false"
-@Dumper.builtin(bool)
class BoolBinaryDumper(Dumper):
format = Format.BINARY
return b"\x01" if obj else b"\x00"
-@Dumper.builtin(type(None))
class NoneDumper(Dumper):
"""
Not a complete dumper as it doesn't implement dump(), but it implements
return b"NULL"
-@Loader.builtin("bool")
class BoolLoader(Loader):
format = Format.TEXT
return data == b"t"
-@Loader.builtin("bool")
class BoolBinaryLoader(Loader):
format = Format.BINARY
from typing import Optional, Union, TYPE_CHECKING
from ..pq import Escaping
-from ..oids import builtins, INVALID_OID
+from ..oids import builtins
from ..adapt import Dumper, Loader, Format
from ..proto import AdaptContext
from ..errors import DataError
self._encoding = enc
-@Dumper.builtin(str)
class StringBinaryDumper(_StringDumper):
format = Format.BINARY
return obj.encode(self._encoding)
-@Dumper.builtin(str)
class StringDumper(_StringDumper):
format = Format.TEXT
return obj.encode(self._encoding)
-@Loader.builtin(INVALID_OID, "bpchar", "name", "text", "varchar")
class TextLoader(Loader):
format = Format.TEXT
return data
-@Loader.builtin("bpchar", "name", "text", "varchar")
class TextBinaryLoader(TextLoader):
format = Format.BINARY
-@Dumper.builtin(bytes, bytearray, memoryview)
class BytesDumper(Dumper):
format = Format.TEXT
return self._esc.escape_bytea(obj)
-@Dumper.builtin(bytes, bytearray, memoryview)
class BytesBinaryDumper(Dumper):
format = Format.BINARY
return obj
-@Loader.builtin("bytea")
class ByteaLoader(Loader):
format = Format.TEXT
return self._escaping.unescape_bytea(data)
-@Loader.builtin("bytea", INVALID_OID)
class ByteaBinaryLoader(Loader):
format = Format.BINARY
UUID: Callable[..., "uuid.UUID"]
-@Dumper.builtin("uuid.UUID")
class UUIDDumper(Dumper):
format = Format.TEXT
return obj.hex.encode("utf8")
-@Dumper.builtin("uuid.UUID")
class UUIDBinaryDumper(UUIDDumper):
format = Format.BINARY
return obj.bytes
-@Loader.builtin("uuid")
class UUIDLoader(Loader):
format = Format.TEXT
return UUID(data.decode("utf8"))
-@Loader.builtin("uuid")
class UUIDBinaryLoader(UUIDLoader):
format = Format.BINARY
@classmethod
def register(
cls,
- int oid,
+ oid: Union[int, str],
context: Optional["AdaptContext"] = None,
int format = Format.TEXT,
) -> None:
+ if isinstance(oid, str):
+ from psycopg3.oids import builtins
+ oid = builtins[oid].oid
+
if context is not None:
adapters = context.adapters
else:
extend-exclude = .venv
per-file-ignores =
./psycopg3/psycopg3/__init__.py: F401
+ ./psycopg3/psycopg3/types/__init__.py: F401
projects / thirdparty / psycopg.git / commitdiff
