from typing_extensions import Protocol
T = TypeVar("T")
-FutureT = Union["asyncio.Future[T]", Generator[Any, None, T], Awaitable[T]]
+FutureT: "TypeAlias" = Union["asyncio.Future[T]", Generator[Any, None, T], Awaitable[T]]
if sys.version_info >= (3, 8):
create_task = asyncio.create_task
from typing import Counter, Deque
if sys.version_info >= (3, 10):
- from typing import TypeGuard
+ from typing import TypeAlias, TypeGuard
else:
- from typing_extensions import TypeGuard
+ from typing_extensions import TypeAlias, TypeGuard
__all__ = [
"Counter",
"Deque",
"Protocol",
+ "TypeAlias",
"TypeGuard",
"ZoneInfo",
"create_task",
from collections import OrderedDict
from .pq import ExecStatus
-from ._compat import Deque
+from ._compat import Deque, TypeAlias
from ._queries import PostgresQuery
if TYPE_CHECKING:
from .pq.abc import PGresult
+Key: TypeAlias = Tuple[bytes, Tuple[int, ...]]
+
class Prepare(IntEnum):
NO = auto()
SHOULD = auto()
-Key = Tuple[bytes, Tuple[int, ...]]
-
-
class PrepareManager:
# Number of times a query is executed before it is prepared.
prepare_threshold: Optional[int] = 5
from typing import Callable, cast, Optional, Tuple
from .abc import Buffer
-from ._compat import Protocol
+from ._compat import Protocol, TypeAlias
-PackInt = Callable[[int], bytes]
-UnpackInt = Callable[[bytes], Tuple[int]]
-PackFloat = Callable[[float], bytes]
-UnpackFloat = Callable[[bytes], Tuple[float]]
+PackInt: TypeAlias = Callable[[int], bytes]
+UnpackInt: TypeAlias = Callable[[bytes], Tuple[int]]
+PackFloat: TypeAlias = Callable[[float], bytes]
+UnpackFloat: TypeAlias = Callable[[bytes], Tuple[float]]
class UnpackLen(Protocol):
from . import errors as e
from .abc import Buffer, LoadFunc, AdaptContext, PyFormat, DumperKey
from .rows import Row, RowMaker
+from ._compat import TypeAlias
from .postgres import INVALID_OID
if TYPE_CHECKING:
from .connection import BaseConnection
NoneType: Type[None] = type(None)
-DumperCache = Dict[DumperKey, "Dumper"]
-OidDumperCache = Dict[int, "Dumper"]
-LoaderCache = Dict[int, "Loader"]
+DumperCache: TypeAlias = Dict[DumperKey, "Dumper"]
+OidDumperCache: TypeAlias = Dict[int, "Dumper"]
+LoaderCache: TypeAlias = Dict[int, "Loader"]
class Transformer(AdaptContext):
from . import errors as e
from .abc import AdaptContext
from .rows import dict_row
+from ._compat import TypeAlias
if TYPE_CHECKING:
from .connection import Connection
from .sql import Identifier
T = TypeVar("T", bound="TypeInfo")
+RegistryKey: TypeAlias = Union[str, int, Tuple[type, int]]
class TypeInfo:
"""
-RegistryKey = Union[str, int, Tuple[type, int]]
-
-
class TypesRegistry:
"""
Container for the information about types in a database.
from . import pq
from ._enums import PyFormat as PyFormat
-from ._compat import Protocol
+from ._compat import Protocol, TypeAlias
if TYPE_CHECKING:
from .sql import Composable
from ._adapters_map import AdaptersMap
# An object implementing the buffer protocol
-Buffer = Union[bytes, bytearray, memoryview]
+Buffer: TypeAlias = Union[bytes, bytearray, memoryview]
-Query = Union[str, bytes, "Composable"]
-Params = Union[Sequence[Any], Mapping[str, Any]]
+Query: TypeAlias = Union[str, bytes, "Composable"]
+Params: TypeAlias = 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, ...]]
+# DumperKey: TypeAlias = Union[type, Tuple[Union[type, "DumperKey"]]]
+DumperKey: TypeAlias = Union[type, Tuple[type, ...]]
# Waiting protocol types
RV = TypeVar("RV")
-PQGenConn = Generator[Tuple[int, "Wait"], "Ready", RV]
+PQGenConn: TypeAlias = Generator[Tuple[int, "Wait"], "Ready", RV]
"""Generator for processes where the connection file number can change.
This can happen in connection and reset, but not in normal querying.
"""
-PQGen = Generator["Wait", "Ready", RV]
+PQGen: TypeAlias = Generator["Wait", "Ready", RV]
"""Generator for processes where the connection file number won't change.
"""
# Adaptation types
-DumpFunc = Callable[[Any], bytes]
-LoadFunc = Callable[[bytes], Any]
+DumpFunc: TypeAlias = Callable[[Any], bytes]
+LoadFunc: TypeAlias = Callable[[bytes], Any]
class AdaptContext(Protocol):
from .adapt import AdaptersMap
from ._enums import IsolationLevel
from .cursor import Cursor
+from ._compat import TypeAlias
from ._cmodule import _psycopg
from .conninfo import make_conninfo, conninfo_to_dict, ConnectionInfo
from .generators import notifies
Notify.__module__ = "psycopg"
-NoticeHandler = Callable[[e.Diagnostic], None]
-NotifyHandler = Callable[[Notify], None]
+NoticeHandler: TypeAlias = Callable[[e.Diagnostic], None]
+NotifyHandler: TypeAlias = Callable[[Notify], None]
class BaseConnection(Generic[Row]):
from contextlib import asynccontextmanager
from . import errors as e
-
from .abc import Query, Params
from .copy import AsyncCopy
from .rows import Row, RowMaker, AsyncRowFactory
from .pq.abc import PGconn, PGresult
from .pq._enums import DiagnosticField
-from ._compat import TypeGuard
+from ._compat import TypeAlias, TypeGuard
-ErrorInfo = Union[None, PGresult, Dict[int, Optional[bytes]]]
+ErrorInfo: TypeAlias = Union[None, PGresult, Dict[int, Optional[bytes]]]
_sqlcodes: Dict[str, "Type[Error]"] = {}
from typing import Union, TYPE_CHECKING
from ._enums import Format, Trace
-from .._compat import Protocol
+from .._compat import Protocol, TypeAlias
if TYPE_CHECKING:
from .misc import PGnotify, ConninfoOption, PGresAttDesc
# An object implementing the buffer protocol (ish)
-Buffer = Union[bytes, bytearray, memoryview]
+Buffer: TypeAlias = Union[bytes, bytearray, memoryview]
class PGconn(Protocol):
from collections import namedtuple
from . import errors as e
-from ._compat import Protocol
+from ._compat import Protocol, TypeAlias
if TYPE_CHECKING:
from .cursor import BaseCursor, Cursor
...
-TupleRow = Tuple[Any, ...]
+TupleRow: TypeAlias = Tuple[Any, ...]
"""
An alias for the type returned by `tuple_row()` (i.e. a tuple of any content).
"""
-DictRow = Dict[str, Any]
+DictRow: TypeAlias = Dict[str, Any]
"""
An alias for the type returned by `dict_row()`
from .. import postgres
from ..abc import Buffer, AdaptContext
from ..adapt import PyFormat, RecursiveDumper, RecursiveLoader
+from .._compat import TypeAlias
from ..postgres import TEXT_OID
from .._typeinfo import TypeInfo
)
-Hstore = Dict[str, Optional[str]]
+Hstore: TypeAlias = Dict[str, Optional[str]]
class BaseHstoreDumper(RecursiveDumper):
from ..pq import Format
from ..abc import AdaptContext
from ..adapt import Buffer, Dumper, Loader
+from .._compat import TypeAlias
if TYPE_CHECKING:
import ipaddress
-Address = Union["ipaddress.IPv4Address", "ipaddress.IPv6Address"]
-Interface = Union["ipaddress.IPv4Interface", "ipaddress.IPv6Interface"]
-Network = Union["ipaddress.IPv4Network", "ipaddress.IPv6Network"]
+Address: TypeAlias = Union["ipaddress.IPv4Address", "ipaddress.IPv6Address"]
+Interface: TypeAlias = Union["ipaddress.IPv4Interface", "ipaddress.IPv6Interface"]
+Network: TypeAlias = Union["ipaddress.IPv4Network", "ipaddress.IPv6Network"]
# These objects will be imported lazily
ip_address: Callable[[str], Address] = None # type: ignore[assignment]
zip_safe = False
install_requires =
backports.zoneinfo >= 0.2.0; python_version < "3.9"
- typing_extensions >= 3.10; python_version < "3.10"
+ typing-extensions >= 3.10; python_version < "3.10"
tzdata; sys_platform == "win32"
[options.package_data]
],
# Requirements to run the test suite
"test": [
- "mypy >= 0.920, != 0.930, != 0.931",
+ "mypy >= 0.940",
"pproxy >= 2.7",
"pytest >= 6.2.5",
"pytest-asyncio >= 0.17",
"black >= 22.1.0",
"dnspython >= 2.1",
"flake8 >= 4.0",
- "mypy >= 0.920, != 0.930, != 0.931",
+ "mypy >= 0.940",
"types-setuptools >= 57.4",
"wheel >= 0.37",
],
import psycopg.errors as e
T = TypeVar("T")
-FutureT = Union["asyncio.Future[T]", Generator[Any, None, T], Awaitable[T]]
+FutureT: "TypeAlias" = Union["asyncio.Future[T]", Generator[Any, None, T], Awaitable[T]]
if sys.version_info >= (3, 8):
create_task = asyncio.create_task
else:
from typing import Counter, Deque
+if sys.version_info >= (3, 10):
+ from typing import TypeAlias
+else:
+ from typing_extensions import TypeAlias
+
__all__ = [
"Counter",
"Deque",
"Task",
+ "TypeAlias",
"create_task",
]
python_requires = >= 3.7
packages = find:
zip_safe = False
-# Maybe it can be useful after release, now it gets in the way.
-# install_requires =
-# psycopg >= 3, < 4
+install_requires =
+ typing-extensions >= 3.10; python_version < "3.10"
[options.package_data]
psycopg_pool = py.typed
# From install_requires
backports.zoneinfo == 0.2.0
-typing_extensions == 3.10.0.0
+typing-extensions == 3.10.0.0
# From the 'test' extra
-mypy == 0.920
+mypy == 0.940
pproxy == 2.7.0
pytest == 6.2.5
pytest-asyncio == 0.17.0
black == 22.1.0
dnspython == 2.1.0
flake8 == 4.0.0
-mypy == 0.920
+mypy == 0.940
types-setuptools == 57.4.0
wheel == 0.37
projects / thirdparty / psycopg.git / commitdiff
