f"dumpers should be registered on classes, got {src} instead"
)
- if not (isinstance(dumper, type) and issubclass(dumper, Dumper)):
+ if not (
+ isinstance(dumper, type) and issubclass(dumper, _dumper_classes)
+ ):
raise TypeError(
f"dumpers should be Dumper subclasses, got {dumper} instead"
)
) -> LoaderType:
if not isinstance(oid, int):
raise TypeError(
- f"typeloaders should be registered on oid, got {oid} instead"
+ f"loaders should be registered on oid, got {oid} instead"
)
- if not (isinstance(loader, type) and issubclass(loader, Loader)):
+ if not (
+ isinstance(loader, type) and issubclass(loader, _loader_classes)
+ ):
raise TypeError(
- f"dumpers should be Loader subclasses, got {loader} instead"
+ f"loaders should be Loader subclasses, got {loader} instead"
)
where = context.loaders if context is not None else Loader.globals
return binary_
+_dumper_classes = (Dumper,)
+_loader_classes = (Loader,)
+
+
def _connection_from_context(
context: AdaptContext,
) -> Optional[BaseConnection]:
from psycopg3_c import _psycopg3
Transformer = _psycopg3.Transformer
+ _loader_classes = (Loader, _psycopg3.PyxLoader) # type: ignore
else:
from . import transform
cdef void register_text_c_loaders():
logger.debug("registering optimised text c loaders")
- from psycopg3.types import text
- register_c_loader(text.StringLoader.load, load_text, get_context_text)
- register_c_loader(text.UnknownLoader.load, load_text, get_context_text)
- register_c_loader(text.load_bytea_text, load_bytea_text)
- register_c_loader(text.load_bytea_binary, load_bytea_binary)
+ from psycopg3.adapt import Loader
+ from psycopg3.types import builtins
+ Loader.register_binary(builtins['bytea'].oid, BinaryByteaLoader)