for dmap in self._dumpers_maps:
for scls in cls.__mro__:
key = (scls, format)
- if key in dmap:
- self._dumpers_cache[key] = dumper = dmap[key](scls, self)
- return dumper
+ dumper_class = dmap.get(key)
+ if not dumper_class:
+ continue
+
+ self._dumpers_cache[key] = dumper = dumper_class(scls, self)
+ return dumper
+
+ # If the adapter is not found, look for its name as a string
+ for dmap in self._dumpers_maps:
+ for scls in cls.__mro__:
+ fqn = f"{cls.__module__}.{scls.__qualname__}"
+ dumper_class = dmap.get((fqn, format))
+ if dumper_class is None:
+ continue
+
+ key = (scls, format)
+ dmap[key] = dumper_class
+ self._dumpers_cache[key] = dumper = dumper_class(scls, self)
+ return dumper
raise e.ProgrammingError(
f"cannot adapt type {type(obj).__name__}"
# Copyright (C) 2020 The Psycopg Team
-from typing import Any, Callable, Optional, Type
+from typing import Any, Callable, Optional, Type, Union
from . import pq
from . import proto
@classmethod
def register(
cls,
- src: type,
+ src: Union[type, str],
context: AdaptContext = None,
format: Format = Format.TEXT,
) -> None:
- if not isinstance(src, type):
+ if not isinstance(src, (str, type)):
raise TypeError(
f"dumpers should be registered on classes, got {src} instead"
)
where[src, format] = cls
@classmethod
- def register_binary(cls, src: type, context: AdaptContext = None) -> None:
+ def register_binary(
+ cls, src: Union[type, str], context: AdaptContext = None
+ ) -> None:
cls.register(src, context, format=Format.BINARY)
@classmethod
- def text(cls, src: type) -> Callable[[DumperType], DumperType]:
+ def text(cls, src: Union[type, str]) -> Callable[[DumperType], DumperType]:
def text_(dumper: DumperType) -> DumperType:
dumper.register(src)
return dumper
return text_
@classmethod
- def binary(cls, src: type) -> Callable[[DumperType], DumperType]:
+ def binary(
+ cls, src: Union[type, str]
+ ) -> Callable[[DumperType], DumperType]:
def binary_(dumper: DumperType) -> DumperType:
dumper.register_binary(src)
return dumper
DumpFunc = Callable[[Any], bytes]
DumperType = Type["Dumper"]
-DumpersMap = Dict[Tuple[type, Format], DumperType]
+DumpersMap = Dict[Tuple[Union[type, str], Format], DumperType]
LoadFunc = Callable[[bytes], Any]
LoaderType = Type["Loader"]
# Copyright (C) 2020 The Psycopg Team
-# TODO: consiter lazy dumper registration.
-import ipaddress
-from ipaddress import IPv4Address, IPv4Interface, IPv4Network
-from ipaddress import IPv6Address, IPv6Interface, IPv6Network
-
-from typing import cast, Callable, Union
+from typing import Callable, Union, TYPE_CHECKING
from ..oids import builtins
from ..adapt import Dumper, Loader
+from ..proto import AdaptContext
+
+if TYPE_CHECKING:
+ import ipaddress
-Address = Union[IPv4Address, IPv6Address]
-Interface = Union[IPv4Interface, IPv6Interface]
-Network = Union[IPv4Network, IPv6Network]
+Address = Union["ipaddress.IPv4Address", "ipaddress.IPv6Address"]
+Interface = Union["ipaddress.IPv4Interface", "ipaddress.IPv6Interface"]
+Network = Union["ipaddress.IPv4Network", "ipaddress.IPv6Network"]
-# in typeshed these types are commented out
-ip_address = cast(Callable[[str], Address], ipaddress.ip_address)
-ip_interface = cast(Callable[[str], Interface], ipaddress.ip_interface)
-ip_network = cast(Callable[[str], Network], ipaddress.ip_network)
+# These functions will be imported lazily
+ip_address: Callable[[str], Address]
+ip_interface: Callable[[str], Interface]
+ip_network: Callable[[str], Network]
-@Dumper.text(IPv4Address)
-@Dumper.text(IPv6Address)
-@Dumper.text(IPv4Interface)
-@Dumper.text(IPv6Interface)
+@Dumper.text("ipaddress.IPv4Address")
+@Dumper.text("ipaddress.IPv6Address")
+@Dumper.text("ipaddress.IPv4Interface")
+@Dumper.text("ipaddress.IPv6Interface")
class InterfaceDumper(Dumper):
oid = builtins["inet"].oid
return str(obj).encode("utf8")
-@Dumper.text(IPv4Network)
-@Dumper.text(IPv6Network)
+@Dumper.text("ipaddress.IPv4Network")
+@Dumper.text("ipaddress.IPv6Network")
class NetworkDumper(Dumper):
oid = builtins["cidr"].oid
return str(obj).encode("utf8")
+class _LazyIpaddress(Loader):
+ def __init__(self, oid: int, context: AdaptContext = None):
+ super().__init__(oid, context)
+ global ip_address, ip_interface, ip_network
+ from ipaddress import ip_address, ip_interface, ip_network
+
+
@Loader.text(builtins["inet"].oid)
-class InetLoader(Loader):
+class InetLoader(_LazyIpaddress):
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(Loader):
+class CidrLoader(_LazyIpaddress):
def load(self, data: bytes) -> Network:
return ip_network(data.decode("utf8"))
# Copyright (C) 2020 The Psycopg Team
-# TODO: importing uuid is slow. Don't import it at module level.
-# Should implement lazy dumper registration.
-from uuid import UUID
+from typing import Callable, TYPE_CHECKING
from ..oids import builtins
from ..adapt import Dumper, Loader
+from ..proto import AdaptContext
+if TYPE_CHECKING:
+ import uuid
-@Dumper.text(UUID)
+# Importing the uuid module is slow, so import it only on request.
+UUID: Callable[..., "uuid.UUID"]
+
+
+@Dumper.text("uuid.UUID")
class UUIDDumper(Dumper):
oid = builtins["uuid"].oid
- def dump(self, obj: UUID) -> bytes:
+ def dump(self, obj: "uuid.UUID") -> bytes:
return obj.hex.encode("utf8")
-@Dumper.binary(UUID)
+@Dumper.binary("uuid.UUID")
class UUIDBinaryDumper(UUIDDumper):
- def dump(self, obj: UUID) -> bytes:
+ def dump(self, obj: "uuid.UUID") -> bytes:
return obj.bytes
@Loader.text(builtins["uuid"].oid)
class UUIDLoader(Loader):
- def load(self, data: bytes) -> UUID:
+ def __init__(self, oid: int, context: AdaptContext = None):
+ super().__init__(oid, context)
+ global UUID
+ from uuid import UUID
+
+ def load(self, data: bytes) -> "uuid.UUID":
return UUID(data.decode("utf8"))
@Loader.binary(builtins["uuid"].oid)
-class UUIDBinaryLoader(Loader):
- def load(self, data: bytes) -> UUID:
+class UUIDBinaryLoader(UUIDLoader):
+ def load(self, data: bytes) -> "uuid.UUID":
return UUID(bytes=data)
for dmap in self._dumpers_maps:
for scls in cls.__mro__:
key = (scls, format)
- if key in dmap:
- self._dumpers_cache[key] = dumper = dmap[key](scls, self)
- return dumper
+ dumper_class = dmap.get(key)
+ if not dumper_class:
+ continue
+
+ self._dumpers_cache[key] = dumper = dumper_class(scls, self)
+ return dumper
+
+ # If the adapter is not found, look for its name as a string
+ for dmap in self._dumpers_maps:
+ for scls in cls.__mro__:
+ fqn = f"{cls.__module__}.{scls.__qualname__}"
+ dumper_class = dmap.get((fqn, format))
+ if dumper_class is None:
+ continue
+
+ key = (scls, format)
+ dmap[key] = dumper_class
+ self._dumpers_cache[key] = dumper = dumper_class(scls, self)
+ return dumper
raise e.ProgrammingError(
f"cannot adapt type {type(obj).__name__}"
+import os
+import sys
import ipaddress
+import subprocess as sp
import pytest
def binary_check(fmt):
if fmt == Format.BINARY:
pytest.xfail("inet binary not implemented")
+
+
+def test_lazy_load(dsn):
+ script = f"""\
+import sys
+import psycopg3
+
+# In 3.6 it seems already loaded (at least on Travis).
+if sys.version_info >= (3, 7):
+ assert 'ipaddress' not in sys.modules
+
+conn = psycopg3.connect({dsn!r})
+with conn.cursor() as cur:
+ cur.execute("select '127.0.0.1'::inet")
+ cur.fetchone()
+
+conn.close()
+assert 'ipaddress' in sys.modules
+"""
+
+ # TODO: debug this. Importing c module fails on travis in this scenario
+ env = dict(os.environ)
+ env.pop("PSYCOPG3_IMPL", None)
+
+ sp.check_call([sys.executable, "-s", "-c", script], env=env)
+import os
+import sys
from uuid import UUID
+import subprocess as sp
import pytest
val = "12345678123456781234567812345679"
cur.execute("select %s::uuid", (val,))
assert cur.fetchone()[0] == UUID(val)
+
+
+def test_lazy_load(dsn):
+ script = f"""\
+import sys
+import psycopg3
+
+assert 'uuid' not in sys.modules
+
+conn = psycopg3.connect({dsn!r})
+with conn.cursor() as cur:
+ cur.execute("select repeat('1', 32)::uuid")
+ cur.fetchone()
+
+conn.close()
+assert 'uuid' in sys.modules
+"""
+
+ # TODO: debug this. Importing c module fails on travis in this scenario
+ env = dict(os.environ)
+ env.pop("PSYCOPG3_IMPL", None)
+
+ sp.check_call([sys.executable, "-c", script], env=env)
projects / thirdparty / psycopg.git / commitdiff
