Where used, in recursive types, get_key() can be used instead.
# Copyright (C) 2020-2021 The Psycopg Team
-from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
+from typing import Any, Dict, List, Optional, Sequence, Tuple
from typing import DefaultDict, TYPE_CHECKING
from collections import defaultdict
from . import errors as e
from .oids import INVALID_OID
from .rows import Row, RowMaker
-from .proto import LoadFunc, AdaptContext, PyFormat
+from .proto import LoadFunc, AdaptContext, PyFormat, DumperKey
if TYPE_CHECKING:
from .pq.proto import PGresult
from .proto import Dumper
from .connection import BaseConnection
-DumperKey = Union[type, Tuple[type, ...]]
DumperCache = Dict[DumperKey, "Dumper"]
-
-LoaderKey = int
-LoaderCache = Dict[LoaderKey, "Loader"]
+LoaderCache = Dict[int, "Loader"]
class Transformer(AdaptContext):
Params = Union[Sequence[Any], Mapping[str, Any]]
ConnectionType = TypeVar("ConnectionType", bound="BaseConnection[Any]")
+# TODO: make it recursive when mypy will support it
+# DumperKey = Union[type, Tuple[Union[type, "DumperKey"]]]
+DumperKey = Union[type, Tuple[type, ...]]
# Waiting protocol types
class Dumper(Protocol):
format: pq.Format
oid: int
- cls: type
def __init__(self, cls: type, context: Optional[AdaptContext] = None):
...
from .. import errors as e
from ..oids import postgres_types, TEXT_OID, TEXT_ARRAY_OID, INVALID_OID
from ..adapt import RecursiveDumper, RecursiveLoader, PyFormat
-from ..proto import Dumper, AdaptContext, Buffer
+from ..proto import AdaptContext, Buffer, Dumper, DumperKey
from .._struct import pack_len, unpack_len
from .._typeinfo import TypeInfo
self.sub_dumper: Optional[Dumper] = None
self._types = context.adapters.types if context else postgres_types
- def get_key(self, obj: List[Any], format: PyFormat) -> Tuple[type, ...]:
+ def get_key(self, obj: List[Any], format: PyFormat) -> DumperKey:
item = self._find_list_element(obj)
if item is not None:
sd = self._tx.get_dumper(item, format)
- return (self.cls, sd.cls)
+ return (self.cls, sd.get_key(item, format)) # type: ignore
else:
return (self.cls,)
import sys
import struct
from datetime import date, datetime, time, timedelta, timezone
-from typing import Any, Callable, cast, Optional, Tuple, Union, TYPE_CHECKING
+from typing import Any, Callable, cast, Optional, Tuple, TYPE_CHECKING
from ..pq import Format
from .._tz import get_tzinfo
from ..oids import postgres_types as builtins
from ..adapt import Buffer, Dumper, Loader, PyFormat
-from ..proto import AdaptContext
+from ..proto import AdaptContext, DumperKey
from ..errors import InterfaceError, DataError
from .._struct import pack_int4, pack_int8, unpack_int4, unpack_int8
class _BaseTimeDumper(Dumper):
- def get_key(self, obj: time, format: PyFormat) -> Union[type, Tuple[type]]:
+ def get_key(self, obj: time, format: PyFormat) -> DumperKey:
# Use (cls,) to report the need to upgrade to a dumper for timetz (the
# Frankenstein of the data types).
if not obj.tzinfo:
class _BaseDatetimeDumper(Dumper):
- def get_key(
- self, obj: datetime, format: PyFormat
- ) -> Union[type, Tuple[type]]:
+ def get_key(self, obj: datetime, format: PyFormat) -> DumperKey:
# Use (cls,) to report the need to upgrade (downgrade, actually) to a
# dumper for naive timestamp.
if obj.tzinfo:
# Copyright (C) 2020-2021 The Psycopg Team
import re
-from typing import Any, Dict, Generic, Optional, Tuple, TypeVar, Type, Union
+from typing import Any, Dict, Generic, Optional, TypeVar, Type, Union
from typing import cast
from decimal import Decimal
from datetime import date, datetime
from ..pq import Format
from ..oids import postgres_types as builtins, INVALID_OID
from ..adapt import RecursiveDumper, RecursiveLoader, PyFormat
-from ..proto import Dumper, AdaptContext, Buffer
+from ..proto import AdaptContext, Buffer, Dumper, DumperKey
from .._struct import pack_len, unpack_len
from .._typeinfo import RangeInfo as RangeInfo # exported here
self._types = context.adapters.types if context else builtins
self._adapt_format = PyFormat.from_pq(self.format)
- def get_key(
- self, obj: Range[Any], format: PyFormat
- ) -> Union[type, Tuple[type, ...]]:
+ def get_key(self, obj: Range[Any], format: PyFormat) -> DumperKey:
# If we are a subclass whose oid is specified we don't need upgrade
if self.oid != INVALID_OID:
return self.cls
item = self._get_item(obj)
if item is not None:
sd = self._tx.get_dumper(item, self._adapt_format)
- return (self.cls, sd.cls)
+ return (self.cls, sd.get_key(item, format)) # type: ignore
else:
return (self.cls,)
return rv
- cdef object get_key(self, object obj, object format):
+ cpdef object get_key(self, object obj, object format):
return self.cls
- cdef object upgrade(self, object obj, object format):
+ cpdef object upgrade(self, object obj, object format):
return self
@classmethod
from __future__ import annotations
from dataclasses import dataclass
-from typing import Any, Callable, Optional, Sequence, Tuple
+from typing import Any, Callable, Optional, Sequence, Tuple, Union
from psycopg import AnyCursor, Connection, Cursor, ServerCursor, connect
from psycopg import pq
format = pq.Format.TEXT
def __init__(self, cls: type, context: Optional[AdaptContext] = None):
- self.cls = cls
+ self._cls = cls
self.oid = 25 # text
def dump(self, obj: str) -> bytes:
return b"'%s'" % esc.escape_string(value.replace(b"h", b"q"))
def get_key(self, obj: str, format: PyFormat) -> type:
- return self.cls
+ return self._cls
def upgrade(self, obj: str, format: PyFormat) -> "MyStrDumper":
return self
+
+
+# This should be the definition of psycopg.adapt.DumperKey, but mypy doesn't
+# support recursive types. When it will, this statement will give an error
+# (unused type: ignore) so we can fix our definition.
+_DumperKey = Union[type, Tuple[Union[type, "_DumperKey"]]] # type: ignore
projects / thirdparty / psycopg.git / commitdiff
