import threading
from typing import Any, Callable, Coroutine, TYPE_CHECKING
-from typing_extensions import TypeAlias
-
-from ._compat import TypeVar
+from ._compat import TypeAlias, TypeVar
Worker: TypeAlias = threading.Thread
AWorker: TypeAlias = "asyncio.Task[None]"
cache = lru_cache(maxsize=None)
if sys.version_info >= (3, 10):
- from typing import TypeGuard
+ from typing import TypeGuard, TypeAlias
else:
- from typing_extensions import TypeGuard
+ from typing_extensions import TypeGuard, TypeAlias
if sys.version_info >= (3, 11):
from typing import LiteralString, Self
"Deque",
"LiteralString",
"Self",
+ "TypeAlias",
"TypeGuard",
"TypeVar",
"ZoneInfo",
from weakref import ref, ReferenceType
from warnings import warn
from functools import partial
-from typing_extensions import TypeAlias
from . import pq
from . import errors as e
from .rows import Row
from .adapt import AdaptersMap
from ._enums import IsolationLevel
-from ._compat import LiteralString, Self, TypeVar
+from ._compat import LiteralString, Self, TypeAlias, TypeVar
from .pq.misc import connection_summary
from ._pipeline import BasePipeline
from ._encodings import pgconn_encoding
from __future__ import annotations
import os
-from typing import Any
+from typing import TYPE_CHECKING
from functools import lru_cache
from ipaddress import ip_address
from dataclasses import dataclass
-from typing_extensions import TypeAlias
from . import pq
from . import errors as e
+from ._compat import TypeAlias
+
+if TYPE_CHECKING:
+ from typing import Any # noqa: F401
ConnDict: TypeAlias = "dict[str, Any]"
import logging
from types import TracebackType
from typing import Any, List, Optional, Union, Tuple, Type, TYPE_CHECKING
-from typing_extensions import TypeAlias
from . import pq
from . import errors as e
from .abc import PipelineCommand, PQGen
-from ._compat import Deque, Self
+from ._compat import Deque, Self, TypeAlias
from .pq.misc import connection_summary
from ._encodings import pgconn_encoding
from ._preparing import Key, Prepare
from enum import IntEnum, auto
from typing import Iterator, Optional, Sequence, Tuple, TYPE_CHECKING
from collections import OrderedDict
-from typing_extensions import TypeAlias
from . import pq
-from ._compat import Deque
+from ._compat import Deque, TypeAlias
from ._queries import PostgresQuery
if TYPE_CHECKING:
from typing import Any, Dict, List, Optional, Sequence, Tuple
from typing import DefaultDict, TYPE_CHECKING
from collections import defaultdict
-from typing_extensions import TypeAlias
from . import pq
from . import abc
from .abc import Buffer, LoadFunc, AdaptContext, PyFormat, DumperKey, NoneType
from .rows import Row, RowMaker
from ._oids import INVALID_OID, TEXT_OID
+from ._compat import TypeAlias
from ._encodings import conn_encoding
if TYPE_CHECKING:
from typing import Any, Callable, Dict, List, Mapping, Match, NamedTuple, Optional
from typing import Sequence, Tuple, Union, TYPE_CHECKING
from functools import lru_cache
-from typing_extensions import TypeAlias
from . import pq
from . import errors as e
from .sql import Composable
from .abc import Buffer, Query, Params
from ._enums import PyFormat
-from ._compat import TypeGuard
+from ._compat import TypeAlias, TypeGuard
from ._encodings import conn_encoding
if TYPE_CHECKING:
import struct
from typing import Callable, cast, Optional, Protocol, Tuple
-from typing_extensions import TypeAlias
-from .abc import Buffer
from . import errors as e
+from .abc import Buffer
+from ._compat import TypeAlias
PackInt: TypeAlias = Callable[[int], bytes]
UnpackInt: TypeAlias = Callable[[Buffer], Tuple[int]]
from typing import Any, Dict, Iterator, Optional, overload
from typing import Sequence, Tuple, Type, Union, TYPE_CHECKING
-from typing_extensions import TypeAlias
from . import sql
from . import errors as e
from .abc import AdaptContext, Query
from .rows import dict_row
-from ._compat import TypeVar
+from ._compat import TypeAlias, TypeVar
from ._encodings import conn_encoding
if TYPE_CHECKING:
from typing import Any, Callable, Generator, Mapping
from typing import List, Optional, Protocol, Sequence, Tuple, Union
from typing import TYPE_CHECKING
-from typing_extensions import TypeAlias
from . import pq
from ._enums import PyFormat as PyFormat
-from ._compat import LiteralString, TypeVar
+from ._compat import LiteralString, TypeAlias, TypeVar
if TYPE_CHECKING:
from . import sql
from dataclasses import dataclass, field, fields
from typing import Any, Callable, Dict, List, NoReturn, Optional, Sequence, Tuple, Type
from typing import Union, TYPE_CHECKING
-from typing_extensions import TypeAlias
from asyncio import CancelledError
from .pq.abc import PGconn, PGresult
from .pq._enums import ConnStatus, DiagnosticField, PipelineStatus, TransactionStatus
-from ._compat import TypeGuard
+from ._compat import TypeAlias, TypeGuard
if TYPE_CHECKING:
from .pq.misc import PGnotify, ConninfoOption
from typing import Any, Callable, List, Optional, Protocol, Sequence, Tuple
from typing import Union, TYPE_CHECKING
-from typing_extensions import TypeAlias
from ._enums import Format, Trace
+from .._compat import TypeAlias
if TYPE_CHECKING:
from .misc import PGnotify, ConninfoOption, PGresAttDesc
from typing import Any, Callable, Dict, List, Optional, NamedTuple, NoReturn
from typing import TYPE_CHECKING, Protocol, Sequence, Tuple, Type
from collections import namedtuple
-from typing_extensions import TypeAlias
from . import pq
from . import errors as e
-from ._compat import TypeVar
+from ._compat import TypeAlias, TypeVar
from ._encodings import _as_python_identifier
if TYPE_CHECKING:
from enum import Enum
from typing import Any, Dict, Generic, Optional, Mapping, Sequence
from typing import Tuple, Type, Union, cast, TYPE_CHECKING
-from typing_extensions import TypeAlias
from .. import sql
from .. import postgres
from ..pq import Format
from ..abc import AdaptContext, Query
from ..adapt import Buffer, Dumper, Loader
-from .._compat import cache, TypeVar
+from .._compat import cache, TypeAlias, TypeVar
from .._encodings import conn_encoding
from .._typeinfo import TypeInfo
import re
from typing import Dict, List, Optional, Type
-from typing_extensions import TypeAlias
from .. import errors as e
from .. import postgres
from ..abc import Buffer, AdaptContext
from .._oids import TEXT_OID
from ..adapt import PyFormat, RecursiveDumper, RecursiveLoader
-from .._compat import cache
+from .._compat import cache, TypeAlias
from .._typeinfo import TypeInfo
_re_escape = re.compile(r'(["\\])')
# Copyright (C) 2020 The Psycopg Team
from typing import Callable, Optional, Type, Union, TYPE_CHECKING
-from typing_extensions import TypeAlias
from .. import _oids
from ..pq import Format
from ..abc import AdaptContext
from ..adapt import Buffer, Dumper, Loader
+from .._compat import TypeAlias
if TYPE_CHECKING:
import ipaddress
import threading
from typing import Any, Callable, Coroutine, TYPE_CHECKING
-from typing_extensions import TypeAlias
-
-from ._compat import TypeVar
+from ._compat import TypeAlias, TypeVar
logger = logging.getLogger("psycopg.pool")
T = TypeVar("T")
else:
from typing import Counter, Deque
+if sys.version_info >= (3, 10):
+ from typing import TypeAlias
+else:
+ from typing_extensions import TypeAlias
+
if sys.version_info >= (3, 11):
from typing import Self
else:
"Counter",
"Deque",
"Self",
+ "TypeAlias",
"TypeVar",
]
from typing import Any, Awaitable, Callable, Union, TYPE_CHECKING
-from typing_extensions import TypeAlias
-
-from ._compat import TypeVar
+from ._compat import TypeAlias, TypeVar
if TYPE_CHECKING:
from .pool import ConnectionPool
import subprocess as sp
from typing import List
from pathlib import Path
-from typing_extensions import TypeAlias
import psycopg
from psycopg.rows import TupleRow
from psycopg.crdb import CrdbConnection
+from psycopg._compat import TypeAlias
Connection: TypeAlias = psycopg.Connection[TupleRow]