import codecs
from typing import Any, Iterable, List, Optional, Sequence, Tuple
+from .proto import AdaptContext, DumpFunc, DumpersMap, DumperType
+from .proto import LoadFunc, LoadersMap, LoaderType, MaybeOid
from .connection import BaseConnection
-from .utils.typing import AdaptContext, DumpFunc, DumpersMap, DumperType
-from .utils.typing import LoadFunc, LoadersMap, LoaderType, MaybeOid
from . import pq
Format = pq.Format
from . import pq
from . import proto
+from .proto import AdaptContext, DumpersMap, DumperType
+from .proto import LoadersMap, LoaderType
from .cursor import BaseCursor
from .connection import BaseConnection
-from .utils.typing import AdaptContext, DumpersMap, DumperType
-from .utils.typing import LoadersMap, LoaderType
# Part of the module interface (just importing it makes mypy unhappy)
Format = pq.Format
from . import errors as e
from . import cursor
from . import generators
+from . import proto
from .conninfo import make_conninfo
from .waiting import wait, wait_async
-from .utils.typing import DumpersMap, LoadersMap
logger = logging.getLogger(__name__)
def __init__(self, pgconn: pq.proto.PGconn):
self.pgconn = pgconn
self.cursor_factory = cursor.BaseCursor
- self.dumpers: DumpersMap = {}
- self.loaders: LoadersMap = {}
+ self.dumpers: proto.DumpersMap = {}
+ self.loaders: proto.LoadersMap = {}
# name of the postgres encoding (in bytes)
self._pgenc = b""
from . import pq
from . import generators
from . import proto
+from .proto import Query, Params, DumpersMap, LoadersMap
from .utils.queries import PostgresQuery
-from .utils.typing import Query, Params, DumpersMap, LoadersMap
if TYPE_CHECKING:
from .connection import BaseConnection, Connection, AsyncConnection
# Copyright (C) 2020 The Psycopg Team
import codecs
-from typing import Any, Iterable, List, Optional, Sequence, Tuple
-from typing import TYPE_CHECKING
+from typing import Any, Callable, Dict, Iterable, List, Mapping, Optional
+from typing import Sequence, Tuple, Type, Union, TYPE_CHECKING
from typing_extensions import Protocol
-from .utils.typing import AdaptContext, DumpFunc, DumpersMap, DumperType
-from .utils.typing import LoadFunc, LoadersMap, LoaderType, MaybeOid
from . import pq
if TYPE_CHECKING:
from .connection import BaseConnection # noqa
+ from .cursor import BaseCursor # noqa
+ from .adapt import Dumper, Loader # noqa
+# Part of the module interface (just importing it makes mypy unhappy)
Format = pq.Format
+EncodeFunc = Callable[[str], Tuple[bytes, int]]
+DecodeFunc = Callable[[bytes], Tuple[str, int]]
+
+Query = Union[str, bytes]
+Params = Union[Sequence[Any], Mapping[str, Any]]
+
+AdaptContext = Union[None, "BaseConnection", "BaseCursor", "Transformer"]
+
+MaybeOid = Union[Optional[bytes], Tuple[Optional[bytes], int]]
+DumpFunc = Callable[[Any], MaybeOid]
+DumperType = Union[Type["Dumper"], DumpFunc]
+DumpersMap = Dict[Tuple[type, Format], DumperType]
+
+LoadFunc = Callable[[bytes], Any]
+LoaderType = Union[Type["Loader"], LoadFunc]
+LoadersMap = Dict[Tuple[int, Format], LoaderType]
+
+
class Transformer(Protocol):
def __init__(self, context: AdaptContext = None):
...
from . import errors as e
from . import pq
+from .proto import AdaptContext, DumpFunc, DumpersMap, DumperType
+from .proto import LoadFunc, LoadersMap, LoaderType, MaybeOid
from .cursor import BaseCursor
-from .types.oids import builtins, INVALID_OID
from .connection import BaseConnection
-from .utils.typing import AdaptContext, DumpFunc, DumpersMap, DumperType
-from .utils.typing import LoadFunc, LoadersMap, LoaderType, MaybeOid
+from .types.oids import builtins, INVALID_OID
Format = pq.Format
TEXT_OID = builtins["text"].oid
from .. import errors as e
from ..adapt import Format, Dumper, Loader, Transformer
+from ..proto import AdaptContext
from .oids import builtins
-from ..utils.typing import AdaptContext
TEXT_OID = builtins["text"].oid
TEXT_ARRAY_OID = builtins["text"].array_oid
from . import array
from ..adapt import Format, Dumper, Loader, Transformer
+from ..proto import AdaptContext
from .oids import builtins, TypeInfo
-from ..utils.typing import AdaptContext
if TYPE_CHECKING:
from ..connection import Connection, AsyncConnection
# Copyright (C) 2020 The Psycopg Team
import codecs
-from typing import Optional, Tuple, Union, TYPE_CHECKING
+from typing import Optional, Tuple, Union
from ..adapt import Dumper, Loader
-from ..utils.typing import EncodeFunc, DecodeFunc
+from ..proto import AdaptContext, EncodeFunc, DecodeFunc
from ..pq import Escaping
from .oids import builtins, INVALID_OID
-if TYPE_CHECKING:
- from ..utils.typing import AdaptContext
TEXT_OID = builtins["text"].oid
BYTEA_OID = builtins["bytea"].oid
@Dumper.text(str)
@Dumper.binary(str)
class StringDumper(Dumper):
- def __init__(self, src: type, context: "AdaptContext"):
+ def __init__(self, src: type, context: AdaptContext):
super().__init__(src, context)
self._encode: EncodeFunc
decode: Optional[DecodeFunc]
- def __init__(self, oid: int, context: "AdaptContext"):
+ def __init__(self, oid: int, context: AdaptContext):
super().__init__(oid, context)
if self.connection is not None:
@Loader.text(builtins["bpchar"].oid)
@Loader.binary(builtins["bpchar"].oid)
class UnknownLoader(Loader):
- def __init__(self, oid: int, context: "AdaptContext"):
+ def __init__(self, oid: int, context: AdaptContext):
super().__init__(oid, context)
self.decode: DecodeFunc
@Dumper.text(bytes)
class BytesDumper(Dumper):
- def __init__(self, src: type, context: "AdaptContext" = None):
+ def __init__(self, src: type, context: AdaptContext = None):
super().__init__(src, context)
self.esc = Escaping(
self.connection.pgconn if self.connection is not None else None
from .. import errors as e
from ..pq import Format
-from .typing import Query, Params
+from ..proto import Query, Params
if TYPE_CHECKING:
from ..proto import Transformer
+++ /dev/null
-"""
-Additional types for checking
-"""
-
-# Copyright (C) 2020 The Psycopg Team
-
-from typing import Any, Callable, Dict, Mapping, Optional, Sequence, Tuple
-from typing import Type, Union, TYPE_CHECKING
-
-from .. import pq
-
-if TYPE_CHECKING:
- from ..connection import BaseConnection # noqa
- from ..cursor import BaseCursor # noqa
- from ..adapt import Dumper, Loader # noqa
- from ..proto import Transformer # noqa
-
-# Part of the module interface (just importing it makes mypy unhappy)
-Format = pq.Format
-
-
-EncodeFunc = Callable[[str], Tuple[bytes, int]]
-DecodeFunc = Callable[[bytes], Tuple[str, int]]
-
-Query = Union[str, bytes]
-Params = Union[Sequence[Any], Mapping[str, Any]]
-
-AdaptContext = Union[None, "BaseConnection", "BaseCursor", "Transformer"]
-
-MaybeOid = Union[Optional[bytes], Tuple[Optional[bytes], int]]
-DumpFunc = Callable[[Any], MaybeOid]
-DumperType = Union[Type["Dumper"], DumpFunc]
-DumpersMap = Dict[Tuple[type, Format], DumperType]
-
-LoadFunc = Callable[[bytes], Any]
-LoaderType = Union[Type["Loader"], LoadFunc]
-LoadersMap = Dict[Tuple[int, Format], LoaderType]