Tuple,
Union,
)
-from functools import partial
from . import exceptions as exc
from .pq import Format, PGresult
@staticmethod
def register(
cls: type,
- adapter: Optional[AdapterType] = None,
+ adapter: AdapterType,
context: Optional[AdaptContext] = None,
format: Format = Format.TEXT,
) -> AdapterType:
- if adapter is None:
- # used as decorator
- return partial(Adapter.register, cls, format=format)
-
if not isinstance(cls, type):
raise TypeError(
f"adapters should be registered on classes, got {cls} instead"
):
raise TypeError(
f"the context should be a connection or cursor,"
- f" got {type(context).__name__}"
+ f" got {type(context)}"
)
if not (
@staticmethod
def register_binary(
cls: type,
- adapter: Optional[AdapterType] = None,
+ adapter: AdapterType,
context: Optional[AdaptContext] = None,
) -> AdapterType:
return Adapter.register(cls, adapter, context, format=Format.BINARY)
+ @staticmethod
+ def text(cls: type) -> Callable[[Any], Any]:
+ def register_adapter_(adapter: AdapterType) -> AdapterType:
+ Adapter.register(cls, adapter)
+ return adapter
+
+ return register_adapter_
+
+ @staticmethod
+ def binary(cls: type) -> Callable[[Any], Any]:
+ def register_binary_adapter_(adapter: AdapterType) -> AdapterType:
+ Adapter.register_binary(cls, adapter)
+ return adapter
+
+ return register_binary_adapter_
+
class Typecaster:
globals: TypecastersMap = {}
@staticmethod
def register(
oid: int,
- caster: Optional[TypecasterType] = None,
+ caster: TypecasterType,
context: Optional[AdaptContext] = None,
format: Format = Format.TEXT,
) -> TypecasterType:
- if caster is None:
- # used as decorator
- return partial(Typecaster.register, oid, format=format)
-
if not isinstance(oid, int):
raise TypeError(
f"typecasters should be registered on oid, got {oid} instead"
):
raise TypeError(
f"the context should be a connection or cursor,"
- f" got {type(context).__name__}"
+ f" got {type(context)}"
)
if not (
@staticmethod
def register_binary(
oid: int,
- caster: Optional[TypecasterType] = None,
+ caster: TypecasterType,
context: Optional[AdaptContext] = None,
) -> TypecasterType:
return Typecaster.register(oid, caster, context, format=Format.BINARY)
+ @staticmethod
+ def text(oid: int) -> Callable[[Any], Any]:
+ def register_caster_(caster: TypecasterType) -> TypecasterType:
+ Typecaster.register(oid, caster)
+ return caster
+
+ return register_caster_
+
+ @staticmethod
+ def binary(oid: int) -> Callable[[Any], Any]:
+ def register_binary_caster_(caster: TypecasterType) -> TypecasterType:
+ Typecaster.register_binary(oid, caster)
+ return caster
+
+ return register_binary_caster_
+
class Transformer:
"""
else:
raise TypeError(
f"the context should be a connection or cursor,"
- f" got {type(context).__name__}"
+ f" got {type(context)}"
)
# mapping class, fmt -> adaptation function
return Adapter.globals[key]
raise exc.ProgrammingError(
- f"cannot adapt type {cls.__name__} to format {Format(fmt).name}"
+ f"cannot adapt type {cls} to format {Format(fmt).name}"
)
def cast_row(self, result: PGresult, n: int) -> Generator[Any, None, None]:
return Typecaster.globals[INVALID_OID, fmt]
-@Typecaster.register(INVALID_OID)
+@Typecaster.text(INVALID_OID)
class UnknownCaster(Typecaster):
"""
Fallback object to convert unknown types to Python
return self.decode(data)[0]
-@Typecaster.register_binary(INVALID_OID)
+@Typecaster.binary(INVALID_OID)
def cast_unknown(data: bytes) -> bytes:
return data
import codecs
from typing import Optional, Union
-from ..adaptation import Adapter, Typecaster
+from ..adaptation import (
+ Adapter,
+ Typecaster,
+)
from ..connection import BaseConnection
from ..utils.typing import EncodeFunc, DecodeFunc
from .oids import type_oid
-@Adapter.register(str)
-@Adapter.register_binary(str)
+@Adapter.text(str)
+@Adapter.binary(str)
class StringAdapter(Adapter):
def __init__(self, cls: type, conn: BaseConnection):
super().__init__(cls, conn)
return self._encode(obj)[0]
-@Typecaster.register(type_oid["text"])
-@Typecaster.register_binary(type_oid["text"])
+@Typecaster.text(type_oid["text"])
+@Typecaster.binary(type_oid["text"])
class StringCaster(Typecaster):
decode: Optional[DecodeFunc]