Clean up TypeVar naems and mark type aliases

Author: bryanforbes

asyncpg/connect_utils.py

@@ -21,6 +21,7 @@
 import sys
 import time
 import typing
+import typing_extensions
 import urllib.parse
 import warnings
 
@@ -37,33 +38,44 @@
 if typing.TYPE_CHECKING:
     from . import connection
 
-_Connection = typing.TypeVar(
-    '_Connection',
+_ConnectionT = typing.TypeVar(
+    '_ConnectionT',
     bound='connection.Connection[typing.Any]'
 )
-_Protocol = typing.TypeVar('_Protocol', bound='protocol.Protocol[typing.Any]')
-_AsyncProtocol = typing.TypeVar(
-    '_AsyncProtocol', bound='asyncio.protocols.Protocol'
+_ProtocolT = typing.TypeVar(
+    '_ProtocolT',
+    bound='protocol.Protocol[typing.Any]'
 )
-_Record = typing.TypeVar('_Record', bound=protocol.Record)
-_SSLMode = typing.TypeVar('_SSLMode', bound='SSLMode')
+_AsyncProtocolT = typing.TypeVar(
+    '_AsyncProtocolT', bound='asyncio.protocols.Protocol'
+)
+_RecordT = typing.TypeVar('_RecordT', bound=protocol.Record)
+_SSLModeT = typing.TypeVar('_SSLModeT', bound='SSLMode')
 
-_TPTupleType = typing.Tuple[asyncio.WriteTransport, _AsyncProtocol]
-AddrType = typing.Union[typing.Tuple[str, int], str]
-SSLStringValues = compat.Literal[
+_TPTupleType: typing_extensions.TypeAlias = typing.Tuple[
+    asyncio.WriteTransport,
+    _AsyncProtocolT
+]
+_SSLStringValues = compat.Literal[
     'disable', 'prefer', 'allow', 'require', 'verify-ca', 'verify-full'
 ]
-_ParsedSSLType = typing.Union[
+AddrType: typing_extensions.TypeAlias = typing.Union[
+    typing.Tuple[str, int],
+    str
+]
+_ParsedSSLType: typing_extensions.TypeAlias = typing.Union[
     ssl_module.SSLContext, compat.Literal[False]
 ]
-SSLType = typing.Union[_ParsedSSLType, SSLStringValues, bool]
-HostType = typing.Union[typing.List[str], str]
-PortListType = typing.Union[
+SSLType: typing_extensions.TypeAlias = typing.Union[
+    _ParsedSSLType, _SSLStringValues, bool
+]
+HostType: typing_extensions.TypeAlias = typing.Union[typing.List[str], str]
+PortListType: typing_extensions.TypeAlias = typing.Union[
     typing.List[typing.Union[int, str]],
     typing.List[int],
     typing.List[str],
 ]
-PortType = typing.Union[
+PortType: typing_extensions.TypeAlias = typing.Union[
     PortListType,
     int,
     str
@@ -80,13 +92,13 @@ class SSLMode(enum.IntEnum):
 
     @classmethod
     def parse(
-        cls: typing.Type[_SSLMode],
-        sslmode: typing.Union[str, _SSLMode]
-    ) -> _SSLMode:
+        cls: typing.Type[_SSLModeT],
+        sslmode: typing.Union[str, _SSLModeT]
+    ) -> _SSLModeT:
         if isinstance(sslmode, cls):
             return sslmode
         return typing.cast(
-            _SSLMode,
+            _SSLModeT,
             getattr(cls, typing.cast(str, sslmode).replace('-', '_'))
         )
 
@@ -798,14 +810,14 @@ def connection_lost(self, exc: typing.Optional[Exception]) -> None:
 
 @typing.overload
 async def _create_ssl_connection(
-    protocol_factory: typing.Callable[[], _Protocol],
+    protocol_factory: typing.Callable[[], _ProtocolT],
     host: str,
     port: int,
     *,
     loop: asyncio.AbstractEventLoop,
     ssl_context: ssl_module.SSLContext,
     ssl_is_advisory: typing.Optional[bool] = False
-) -> _TPTupleType[_Protocol]:
+) -> _TPTupleType[_ProtocolT]:
     ...
 
 
@@ -824,7 +836,7 @@ async def _create_ssl_connection(
 
 async def _create_ssl_connection(
     protocol_factory: typing.Union[
-        typing.Callable[[], _Protocol],
+        typing.Callable[[], _ProtocolT],
         typing.Callable[[], '_CancelProto']
     ],
     host: str,
@@ -886,7 +898,7 @@ async def _create_ssl_connection(
 
         try:
             new_tr, pg_proto = typing.cast(
-                typing.Tuple[asyncio.WriteTransport, _Protocol],
+                typing.Tuple[asyncio.WriteTransport, _ProtocolT],
                 await conn_factory(sock=sock)
             )
             pg_proto.is_ssl = do_ssl_upgrade
@@ -903,9 +915,9 @@ async def _connect_addr(
     timeout: float,
     params: _ConnectionParameters,
     config: _ClientConfiguration,
-    connection_class: typing.Type[_Connection],
-    record_class: typing.Type[_Record]
-) -> _Connection:
+    connection_class: typing.Type[_ConnectionT],
+    record_class: typing.Type[_RecordT]
+) -> _ConnectionT:
     assert loop is not None
 
     if timeout <= 0:
@@ -956,22 +968,22 @@ async def __connect_addr(
     addr: AddrType,
     loop: asyncio.AbstractEventLoop,
     config: _ClientConfiguration,
-    connection_class: typing.Type[_Connection],
-    record_class: typing.Type[_Record],
+    connection_class: typing.Type[_ConnectionT],
+    record_class: typing.Type[_RecordT],
     params_input: _ConnectionParameters,
-) -> _Connection:
+) -> _ConnectionT:
     connected = _create_future(loop)
 
     proto_factory: typing.Callable[
-        [], 'protocol.Protocol[_Record]'
+        [], 'protocol.Protocol[_RecordT]'
     ] = lambda: protocol.Protocol(
         addr, connected, params, record_class, loop)
 
     if isinstance(addr, str):
         # UNIX socket
         connector = typing.cast(
             typing.Coroutine[
-                typing.Any, None, _TPTupleType['protocol.Protocol[_Record]']
+                typing.Any, None, _TPTupleType['protocol.Protocol[_RecordT]']
             ],
             loop.create_unix_connection(proto_factory, addr)
         )
@@ -981,7 +993,7 @@ async def __connect_addr(
         # SSL connection
         connector = typing.cast(
             typing.Coroutine[
-                typing.Any, None, _TPTupleType['protocol.Protocol[_Record]']
+                typing.Any, None, _TPTupleType['protocol.Protocol[_RecordT]']
             ],
             loop.create_connection(
                 proto_factory, *addr, ssl=params.ssl
@@ -995,7 +1007,7 @@ async def __connect_addr(
     else:
         connector = typing.cast(
             typing.Coroutine[
-                typing.Any, None, _TPTupleType['protocol.Protocol[_Record]']
+                typing.Any, None, _TPTupleType['protocol.Protocol[_RecordT]']
             ],
             loop.create_connection(proto_factory, *addr)
         )
@@ -1053,10 +1065,10 @@ async def _connect(
     *,
     loop: typing.Optional[asyncio.AbstractEventLoop],
     timeout: float,
-    connection_class: typing.Type[_Connection],
-    record_class: typing.Type[_Record],
+    connection_class: typing.Type[_ConnectionT],
+    record_class: typing.Type[_RecordT],
     **kwargs: typing.Any
-) -> _Connection:
+) -> _ConnectionT:
     if loop is None:
         loop = asyncio.get_event_loop()