Because I have ArrayCaster, will have RangeCaster etc.
AdapterType = Union["Adapter", AdapterFunc]
AdaptersMap = Dict[Tuple[type, Format], AdapterType]
-TypecasterFunc = Callable[[bytes], Any]
-TypecasterType = Union["Typecaster", TypecasterFunc]
-TypecastersMap = Dict[Tuple[int, Format], TypecasterType]
+TypeCasterFunc = Callable[[bytes], Any]
+TypeCasterType = Union["TypeCaster", TypeCasterFunc]
+TypeCastersMap = Dict[Tuple[int, Format], TypeCasterType]
class Adapter:
return binary_
-class Typecaster:
- globals: TypecastersMap = {}
+class TypeCaster:
+ globals: TypeCastersMap = {}
def __init__(self, oid: int, conn: Optional[BaseConnection]):
self.oid = oid
@staticmethod
def register(
oid: int,
- caster: TypecasterType,
+ caster: TypeCasterType,
context: AdaptContext = None,
format: Format = Format.TEXT,
- ) -> TypecasterType:
+ ) -> TypeCasterType:
if not isinstance(oid, int):
raise TypeError(
f"typecasters should be registered on oid, got {oid} instead"
if not (
callable(caster)
- or (isinstance(caster, type) and issubclass(caster, Typecaster))
+ or (isinstance(caster, type) and issubclass(caster, TypeCaster))
):
raise TypeError(
- f"adapters should be callable or Typecaster subclasses,"
+ f"adapters should be callable or TypeCaster subclasses,"
f" got {caster} instead"
)
- where = context.casters if context is not None else Typecaster.globals
+ where = context.casters if context is not None else TypeCaster.globals
where[oid, format] = caster
return caster
@staticmethod
def register_binary(
- oid: int, caster: TypecasterType, context: AdaptContext = None,
- ) -> TypecasterType:
- return Typecaster.register(oid, caster, context, format=Format.BINARY)
+ oid: int, caster: TypeCasterType, context: AdaptContext = None,
+ ) -> TypeCasterType:
+ return TypeCaster.register(oid, caster, context, format=Format.BINARY)
@staticmethod
def text(oid: int) -> Callable[[Any], Any]:
- def text_(caster: TypecasterType) -> TypecasterType:
- Typecaster.register(oid, caster)
+ def text_(caster: TypeCasterType) -> TypeCasterType:
+ TypeCaster.register(oid, caster)
return caster
return text_
@staticmethod
def binary(oid: int) -> Callable[[Any], Any]:
- def binary_(caster: TypecasterType) -> TypecasterType:
- Typecaster.register_binary(oid, caster)
+ def binary_(caster: TypeCasterType) -> TypeCasterType:
+ TypeCaster.register_binary(oid, caster)
return caster
return binary_
self._adapt_funcs: Dict[Tuple[type, Format], AdapterFunc] = {}
# mapping oid, fmt -> cast function
- self._cast_funcs: Dict[Tuple[int, Format], TypecasterFunc] = {}
+ self._cast_funcs: Dict[Tuple[int, Format], TypeCasterFunc] = {}
# The result to return values from
self._result: Optional[PGresult] = None
# sequence of cast function from value to python
# the length of the result columns
- self._row_casters: List[TypecasterFunc] = []
+ self._row_casters: List[TypeCasterFunc] = []
@property
def result(self) -> Optional[PGresult]:
else:
return None
- def get_cast_function(self, oid: int, fmt: Format) -> TypecasterFunc:
+ def get_cast_function(self, oid: int, fmt: Format) -> TypeCasterFunc:
try:
return self._cast_funcs[oid, fmt]
except KeyError:
if isinstance(caster, type):
return caster(oid, self.connection).cast
else:
- return cast(TypecasterFunc, caster)
+ return cast(TypeCasterFunc, caster)
- def lookup_caster(self, oid: int, fmt: Format) -> TypecasterType:
+ def lookup_caster(self, oid: int, fmt: Format) -> TypeCasterType:
key = (oid, fmt)
cur = self.cursor
if conn is not None and key in conn.casters:
return conn.casters[key]
- if key in Typecaster.globals:
- return Typecaster.globals[key]
+ if key in TypeCaster.globals:
+ return TypeCaster.globals[key]
- return Typecaster.globals[INVALID_OID, fmt]
+ return TypeCaster.globals[INVALID_OID, fmt]
-@Typecaster.text(INVALID_OID)
-class UnknownCaster(Typecaster):
+@TypeCaster.text(INVALID_OID)
+class UnknownCaster(TypeCaster):
"""
Fallback object to convert unknown types to Python
"""
return self.decode(data)[0]
-@Typecaster.binary(INVALID_OID)
+@TypeCaster.binary(INVALID_OID)
def cast_unknown(data: bytes) -> bytes:
return data
RV = TypeVar("RV")
if TYPE_CHECKING:
- from .adapt import AdaptersMap, TypecastersMap
+ from .adapt import AdaptersMap, TypeCastersMap
class BaseConnection:
self.pgconn = pgconn
self.cursor_factory = cursor.BaseCursor
self.adapters: AdaptersMap = {}
- self.casters: TypecastersMap = {}
+ self.casters: TypeCastersMap = {}
# name of the postgres encoding (in bytes)
self._pgenc = b""
AsyncConnection,
QueryGen,
)
- from .adapt import AdaptersMap, TypecastersMap
+ from .adapt import AdaptersMap, TypeCastersMap
class BaseCursor:
self.conn = conn
self.binary = binary
self.adapters: AdaptersMap = {}
- self.casters: TypecastersMap = {}
+ self.casters: TypeCastersMap = {}
self._reset()
def _reset(self) -> None:
from .. import errors as e
from ..pq import Format
-from ..adapt import Adapter, Typecaster, Transformer, UnknownCaster
-from ..adapt import AdaptContext, TypecasterType, TypecasterFunc
+from ..adapt import Adapter, TypeCaster, Transformer, UnknownCaster
+from ..adapt import AdaptContext, TypeCasterType, TypeCasterFunc
if TYPE_CHECKING:
from ..connection import BaseConnection
tokens[-1] = b"}"
-class ArrayCasterBase(Typecaster):
- base_caster: TypecasterType
+class ArrayCasterBase(TypeCaster):
+ base_caster: TypeCasterType
def __init__(
self, oid: int, conn: Optional["BaseConnection"],
):
super().__init__(oid, conn)
- self.caster_func = TypecasterFunc # type: ignore
+ self.caster_func = TypeCasterFunc # type: ignore
if isinstance(self.base_caster, type):
self.caster_func = self.base_caster(oid, conn).cast
else:
- self.caster_func = cast(TypecasterFunc, type(self).base_caster)
+ self.caster_func = cast(TypeCasterFunc, type(self).base_caster)
def cast(self, data: bytes) -> List[Any]:
rv = None
return rv
-class ArrayCaster(Typecaster):
+class ArrayCaster(TypeCaster):
@staticmethod
def register(
oid: int, # array oid
- caster: TypecasterType,
+ caster: TypeCasterType,
context: AdaptContext = None,
format: Format = Format.TEXT,
- ) -> TypecasterType:
+ ) -> TypeCasterType:
t = type(
caster.__name__ + "_array", # type: ignore
(ArrayCasterBase,),
{"base_caster": caster},
)
- return Typecaster.register(oid, t, context=context, format=format)
+ return TypeCaster.register(oid, t, context=context, format=format)
@staticmethod
def text(oid: int) -> Callable[[Any], Any]:
- def text_(caster: TypecasterType) -> TypecasterType:
+ def text_(caster: TypeCasterType) -> TypeCasterType:
ArrayCaster.register(oid, caster, format=Format.TEXT)
return caster
from decimal import Decimal
from typing import Tuple
-from ..adapt import Adapter, Typecaster
+from ..adapt import Adapter, TypeCaster
from .oids import builtins
from .array import ArrayCaster
return _bool_adapt[obj]
-@Typecaster.text(builtins["int2"].oid)
-@Typecaster.text(builtins["int4"].oid)
-@Typecaster.text(builtins["int8"].oid)
-@Typecaster.text(builtins["oid"].oid)
+@TypeCaster.text(builtins["int2"].oid)
+@TypeCaster.text(builtins["int4"].oid)
+@TypeCaster.text(builtins["int8"].oid)
+@TypeCaster.text(builtins["oid"].oid)
@ArrayCaster.text(builtins["int2"].array_oid)
@ArrayCaster.text(builtins["int4"].array_oid)
@ArrayCaster.text(builtins["int8"].array_oid)
return int(_decode(data)[0])
-@Typecaster.binary(builtins["int2"].oid)
+@TypeCaster.binary(builtins["int2"].oid)
def cast_binary_int2(data: bytes) -> int:
rv: int = _int2_struct.unpack(data)[0]
return rv
-@Typecaster.binary(builtins["int4"].oid)
+@TypeCaster.binary(builtins["int4"].oid)
def cast_binary_int4(data: bytes) -> int:
rv: int = _int4_struct.unpack(data)[0]
return rv
-@Typecaster.binary(builtins["int8"].oid)
+@TypeCaster.binary(builtins["int8"].oid)
def cast_binary_int8(data: bytes) -> int:
rv: int = _int8_struct.unpack(data)[0]
return rv
-@Typecaster.binary(builtins["oid"].oid)
+@TypeCaster.binary(builtins["oid"].oid)
def cast_binary_oid(data: bytes) -> int:
rv: int = _oid_struct.unpack(data)[0]
return rv
-@Typecaster.text(builtins["float4"].oid)
-@Typecaster.text(builtins["float8"].oid)
+@TypeCaster.text(builtins["float4"].oid)
+@TypeCaster.text(builtins["float8"].oid)
def cast_float(data: bytes) -> float:
# it supports bytes directly
return float(data)
-@Typecaster.binary(builtins["float4"].oid)
+@TypeCaster.binary(builtins["float4"].oid)
def cast_binary_float4(data: bytes) -> float:
rv: float = _float4_struct.unpack(data)[0]
return rv
-@Typecaster.binary(builtins["float8"].oid)
+@TypeCaster.binary(builtins["float8"].oid)
def cast_binary_float8(data: bytes) -> float:
rv: float = _float8_struct.unpack(data)[0]
return rv
-@Typecaster.text(builtins["numeric"].oid)
+@TypeCaster.text(builtins["numeric"].oid)
def cast_numeric(data: bytes) -> Decimal:
return Decimal(_decode(data)[0])
_bool_binary_casts = {b"\x01": True, b"\x00": False}
-@Typecaster.text(builtins["bool"].oid)
+@TypeCaster.text(builtins["bool"].oid)
def cast_bool(data: bytes) -> bool:
return _bool_casts[data]
-@Typecaster.binary(builtins["bool"].oid)
+@TypeCaster.binary(builtins["bool"].oid)
def cast_binary_bool(data: bytes) -> bool:
return _bool_binary_casts[data]
from ..adapt import (
Adapter,
- Typecaster,
+ TypeCaster,
)
from ..connection import BaseConnection
from ..utils.typing import EncodeFunc, DecodeFunc
return self._encode(obj)[0]
-@Typecaster.text(builtins["text"].oid)
-@Typecaster.binary(builtins["text"].oid)
+@TypeCaster.text(builtins["text"].oid)
+@TypeCaster.binary(builtins["text"].oid)
@ArrayCaster.text(builtins["text"].array_oid)
-class StringCaster(Typecaster):
+class StringCaster(TypeCaster):
decode: Optional[DecodeFunc]
return b, BYTEA_OID
-@Typecaster.text(builtins["bytea"].oid)
+@TypeCaster.text(builtins["bytea"].oid)
@ArrayCaster.text(builtins["bytea"].array_oid)
def cast_bytea(data: bytes) -> bytes:
return Escaping.unescape_bytea(data)
-@Typecaster.binary(builtins["bytea"].oid)
+@TypeCaster.binary(builtins["bytea"].oid)
def cast_bytea_binary(data: bytes) -> bytes:
return data
import pytest
-from psycopg3.adapt import Transformer, Format, Adapter, Typecaster
+from psycopg3.adapt import Transformer, Format, Adapter, TypeCaster
from psycopg3.types.oids import builtins
TEXT_OID = builtins["text"].oid
def test_cast_connection_ctx(conn):
- Typecaster.register(TEXT_OID, lambda b: b.decode("ascii") + "t", conn)
- Typecaster.register_binary(
+ TypeCaster.register(TEXT_OID, lambda b: b.decode("ascii") + "t", conn)
+ TypeCaster.register_binary(
TEXT_OID, lambda b: b.decode("ascii") + "b", conn
)
def test_cast_cursor_ctx(conn):
- Typecaster.register(TEXT_OID, lambda b: b.decode("ascii") + "t", conn)
- Typecaster.register_binary(
+ TypeCaster.register(TEXT_OID, lambda b: b.decode("ascii") + "t", conn)
+ TypeCaster.register_binary(
TEXT_OID, lambda b: b.decode("ascii") + "b", conn
)
cur = conn.cursor()
- Typecaster.register(TEXT_OID, lambda b: b.decode("ascii") + "tc", cur)
- Typecaster.register_binary(
+ TypeCaster.register(TEXT_OID, lambda b: b.decode("ascii") + "tc", cur)
+ TypeCaster.register_binary(
TEXT_OID, lambda b: b.decode("ascii") + "bc", cur
)
import pytest
from psycopg3.types import builtins
-from psycopg3.adapt import Typecaster, UnknownCaster
+from psycopg3.adapt import TypeCaster, UnknownCaster
from psycopg3.types.array import UnknownArrayCaster, ArrayCaster
def test_unknown(conn):
# unknown for real
- assert builtins["aclitem"].array_oid not in Typecaster.globals
- Typecaster.register(
+ assert builtins["aclitem"].array_oid not in TypeCaster.globals
+ TypeCaster.register(
builtins["aclitem"].array_oid, UnknownArrayCaster, context=conn
)
cur = conn.cursor()
import pytest
-from psycopg3.adapt import Typecaster, Format
+from psycopg3.adapt import TypeCaster, Format
from psycopg3.types import builtins
from psycopg3.types.numeric import cast_float
)
def test_numeric_as_float(conn, val):
cur = conn.cursor()
- Typecaster.register(builtins["numeric"].oid, cast_float, cur)
+ TypeCaster.register(builtins["numeric"].oid, cast_float, cur)
val = Decimal(val)
cur.execute("select %s", (val,))