Remove TypeAlias annotations and handle Literal properly

Author: bryanforbes

asyncpg/compat.py

@@ -13,13 +13,11 @@
 
 if sys.version_info >= (3, 8):
     from typing import (
-        Literal as Literal,
         Protocol as Protocol,
         TypedDict as TypedDict
     )
 else:
     from typing_extensions import (  # noqa: F401
-        Literal as Literal,
         Protocol as Protocol,
         TypedDict as TypedDict
     )

asyncpg/connect_utils.py

@@ -21,15 +21,14 @@
 import sys
 import time
 import typing
-import typing_extensions
 import urllib.parse
 import warnings
 
 # Work around https://github.com/microsoft/pyright/issues/3012
 if sys.version_info >= (3, 8):
-    from typing import Final
+    from typing import Final, Literal
 else:
-    from typing_extensions import Final
+    from typing_extensions import Final, Literal
 
 from . import compat
 from . import exceptions
@@ -52,30 +51,30 @@
 _RecordT = typing.TypeVar('_RecordT', bound=protocol.Record)
 _SSLModeT = typing.TypeVar('_SSLModeT', bound='SSLMode')
 
-_TPTupleType: typing_extensions.TypeAlias = typing.Tuple[
+_TPTupleType = typing.Tuple[
     asyncio.WriteTransport,
     _AsyncProtocolT
 ]
-_SSLStringValues = compat.Literal[
+_SSLStringValues = Literal[
     'disable', 'prefer', 'allow', 'require', 'verify-ca', 'verify-full'
 ]
-AddrType: typing_extensions.TypeAlias = typing.Union[
+AddrType = typing.Union[
     typing.Tuple[str, int],
     str
 ]
-_ParsedSSLType: typing_extensions.TypeAlias = typing.Union[
-    ssl_module.SSLContext, compat.Literal[False]
+_ParsedSSLType = typing.Union[
+    ssl_module.SSLContext, Literal[False]
 ]
-SSLType: typing_extensions.TypeAlias = typing.Union[
+SSLType = typing.Union[
     _ParsedSSLType, _SSLStringValues, bool
 ]
-HostType: typing_extensions.TypeAlias = typing.Union[typing.List[str], str]
-PortListType: typing_extensions.TypeAlias = typing.Union[
+HostType = typing.Union[typing.List[str], str]
+PortListType = typing.Union[
     typing.List[typing.Union[int, str]],
     typing.List[int],
     typing.List[str],
 ]
-PortType: typing_extensions.TypeAlias = typing.Union[
+PortType = typing.Union[
     PortListType,
     int,
     str

asyncpg/connection.py

@@ -6,7 +6,6 @@
 
 
 import asyncio
-import typing_extensions
 import asyncpg
 import collections
 import collections.abc
@@ -33,40 +32,45 @@
 from . import types
 from . import utils
 
+# Work around https://github.com/microsoft/pyright/issues/3012
+if sys.version_info >= (3, 8):
+    from typing import Literal
+else:
+    from typing_extensions import Literal
+
 if typing.TYPE_CHECKING:
     import io
     from .protocol import protocol as _cprotocol
     from . import pool_connection_proxy as _pool
 
-
 _ConnectionT = typing.TypeVar('_ConnectionT', bound='Connection[typing.Any]')
 _RecordT = typing.TypeVar('_RecordT', bound=protocol.Record)
 _OtherRecordT = typing.TypeVar('_OtherRecordT', bound=protocol.Record)
 
-_Writer: typing_extensions.TypeAlias = typing.Callable[
+_Writer = typing.Callable[
     [bytes],
     typing.Coroutine[typing.Any, typing.Any, None]
 ]
-_RecordsType: typing_extensions.TypeAlias = typing.List[_RecordT]
-_RecordsExtraType: typing_extensions.TypeAlias = typing.Tuple[
+_RecordsType = typing.List[_RecordT]
+_RecordsExtraType = typing.Tuple[
     _RecordsType[_RecordT],
     bytes,
     bool
 ]
 
-OutputType: typing_extensions.TypeAlias = typing.Union[
+OutputType = typing.Union[
     'os.PathLike[typing.Any]',
     typing.BinaryIO,
     _Writer
 ]
-SourceType: typing_extensions.TypeAlias = typing.Union[
+SourceType = typing.Union[
     'os.PathLike[typing.Any]',
     typing.BinaryIO,
     typing.AsyncIterable[bytes]
 ]
 
-CopyFormat = compat.Literal['text', 'csv', 'binary']
-PasswordType: typing_extensions.TypeAlias = typing.Union[
+CopyFormat = Literal['text', 'csv', 'binary']
+PasswordType = typing.Union[
     str,
     typing.Callable[[], str],
     typing.Callable[[], typing.Awaitable[str]]
@@ -2133,7 +2137,7 @@ async def _execute(
         limit: int,
         timeout: typing.Optional[float],
         *,
-        return_status: compat.Literal[False] = ...,
+        return_status: Literal[False] = ...,
         ignore_custom_codec: bool = ...,
         record_class: None = ...
     ) -> _RecordsType[_RecordT]:
@@ -2147,7 +2151,7 @@ async def _execute(
         limit: int,
         timeout: typing.Optional[float],
         *,
-        return_status: compat.Literal[False] = ...,
+        return_status: Literal[False] = ...,
         ignore_custom_codec: bool = ...,
         record_class: typing.Type[_OtherRecordT]
     ) -> _RecordsType[_OtherRecordT]:
@@ -2161,7 +2165,7 @@ async def _execute(
         limit: int,
         timeout: typing.Optional[float],
         *,
-        return_status: compat.Literal[False] = ...,
+        return_status: Literal[False] = ...,
         ignore_custom_codec: bool = ...,
         record_class: typing.Optional[typing.Type[_OtherRecordT]]
     ) -> typing.Union[_RecordsType[_RecordT], _RecordsType[_OtherRecordT]]:
@@ -2175,7 +2179,7 @@ async def _execute(
         limit: int,
         timeout: typing.Optional[float],
         *,
-        return_status: compat.Literal[True],
+        return_status: Literal[True],
         ignore_custom_codec: bool = ...,
         record_class: None = ...
     ) -> _RecordsExtraType[_RecordT]:
@@ -2189,7 +2193,7 @@ async def _execute(
         limit: int,
         timeout: typing.Optional[float],
         *,
-        return_status: compat.Literal[True],
+        return_status: Literal[True],
         ignore_custom_codec: bool = ...,
         record_class: typing.Type[_OtherRecordT]
     ) -> _RecordsExtraType[_OtherRecordT]:
@@ -2226,7 +2230,7 @@ async def __execute(
         limit: int,
         timeout: typing.Optional[float],
         *,
-        return_status: compat.Literal[False] = ...,
+        return_status: Literal[False] = ...,
         ignore_custom_codec: bool = ...,
         record_class: None = ...
     ) -> typing.Tuple[
@@ -2243,7 +2247,7 @@ async def __execute(
         limit: int,
         timeout: typing.Optional[float],
         *,
-        return_status: compat.Literal[False] = ...,
+        return_status: Literal[False] = ...,
         ignore_custom_codec: bool = ...,
         record_class: typing.Type[_OtherRecordT]
     ) -> typing.Tuple[
@@ -2260,7 +2264,7 @@ async def __execute(
         limit: int,
         timeout: typing.Optional[float],
         *,
-        return_status: compat.Literal[True],
+        return_status: Literal[True],
         ignore_custom_codec: bool = ...,
         record_class: None = ...
     ) -> typing.Tuple[
@@ -2277,7 +2281,7 @@ async def __execute(
         limit: int,
         timeout: typing.Optional[float],
         *,
-        return_status: compat.Literal[True],
+        return_status: Literal[True],
         ignore_custom_codec: bool = ...,
         record_class: typing.Type[_OtherRecordT]
     ) -> typing.Tuple[

asyncpg/protocol/protocol.pyi

@@ -19,16 +19,15 @@ from typing import (
 )
 
 if sys.version_info >= (3, 10):
-    from typing import Final, TypeAlias, final
+    from typing import Final, Literal, TypeAlias, final
 elif sys.version_info >= (3, 8):
-    from typing import Final, final
+    from typing import Final, Literal, final
     from typing_extensions import TypeAlias
 else:
-    from typing_extensions import Final, TypeAlias, final
+    from typing_extensions import Final, Literal, TypeAlias, final
 
 import asyncpg.pgproto.pgproto
 
-from ..compat import Literal
 from ..connect_utils import _ConnectionParameters
 from ..pgproto.pgproto import WriteBuffer
 from ..types import Type, Attribute

asyncpg/transaction.py

@@ -11,11 +11,10 @@
 
 # Work around https://github.com/microsoft/pyright/issues/3012
 if sys.version_info >= (3, 8):
-    from typing import Final
+    from typing import Final, Literal
 else:
-    from typing_extensions import Final
+    from typing_extensions import Final, Literal
 
-from . import compat
 from . import connresource
 from . import exceptions as apg_errors
 
@@ -32,9 +31,11 @@ class TransactionState(enum.Enum):
     FAILED = 4
 
 
-IsolationLevels = compat.Literal['read_committed',
-                                 'serializable',
-                                 'repeatable_read']
+IsolationLevels = Literal[
+    'read_committed',
+    'serializable',
+    'repeatable_read'
+]
 ISOLATION_LEVELS: Final[
     typing.Set[IsolationLevels]
 ] = {