[Azure AMQP] Typing (#33299)

* init conn typing

* fix async

* fix mypy

* pylint fixes for conn

* type encode/decode

* type auth

* link typing

* remove bracket

* minor fixes

* typing for cbs auth

* minor fixes

* minor fixes

* add in defaults

* pylint fixes

* fix cyclic dependency

* clean up auth type ignores

* clean up leftover py2 comments

* fix type checking

* fix initial comments

* add type to dict

* additional feedback fixes

* remove some casts

* remove unused import

* rebased changes

* fix tests

* fix tests + fix typing

* pylint

* async auth separate class

* fix connection tests

* remove casts

* address comments

* bring in changes to EH vendored version

* fix import

* remove cast from cbs

* add scheme to fake url

* fix bug

* bring in changes in to SB

* restore eh clients

* fix pylint

* EH async tests

* address PR comments

* reset frame typing back

* remove nametuple typing

* additional typing for known amqp types

* last couple of typing changes

Author: kashifkhan

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_connection.py

@@ -8,7 +8,19 @@
 import struct
 import uuid
 import logging
-from typing import List, Optional, Tuple, Dict, Callable, Any, cast, Union, TYPE_CHECKING  # pylint: disable=unused-import
+from typing import (
+    Callable,
+    List,
+    Optional,
+    Tuple,
+    Dict,
+    Any,
+    Union,
+    cast,
+    TYPE_CHECKING,
+)
+
+from typing_extensions import Literal
 
 
 from .message import Message, Header, Properties
@@ -39,130 +51,105 @@
 c_double = struct.Struct('>d')
 
 
-def _decode_null(buffer):
-    # type: (memoryview) -> Tuple[memoryview, None]
+def _decode_null(buffer: memoryview) -> Tuple[memoryview, None]:
     return buffer, None
 
 
-def _decode_true(buffer):
-    # type: (memoryview) -> Tuple[memoryview, bool]
+def _decode_true(buffer: memoryview) -> Tuple[memoryview, Literal[True]]:
     return buffer, True
 
 
-def _decode_false(buffer):
-    # type: (memoryview) -> Tuple[memoryview, bool]
+def _decode_false(buffer: memoryview) -> Tuple[memoryview, Literal[False]]:
     return buffer, False
 
 
-def _decode_zero(buffer):
-    # type: (memoryview) -> Tuple[memoryview, int]
+def _decode_zero(buffer: memoryview) -> Tuple[memoryview, Literal[0]]:
     return buffer, 0
 
 
-def _decode_empty(buffer):
-    # type: (memoryview) -> Tuple[memoryview, List[None]]
+def _decode_empty(buffer: memoryview) -> Tuple[memoryview, List[Any]]:
     return buffer, []
 
 
-def _decode_boolean(buffer):
-    # type: (memoryview) -> Tuple[memoryview, bool]
+def _decode_boolean(buffer: memoryview) -> Tuple[memoryview, bool]:
     return buffer[1:], buffer[:1] == b'\x01'
 
 
-def _decode_ubyte(buffer):
-    # type: (memoryview) -> Tuple[memoryview, int]
+def _decode_ubyte(buffer: memoryview) -> Tuple[memoryview, int]:
     return buffer[1:], buffer[0]
 
 
-def _decode_ushort(buffer):
-    # type: (memoryview) -> Tuple[memoryview, int]
+def _decode_ushort(buffer: memoryview) -> Tuple[memoryview, int]:
     return buffer[2:], c_unsigned_short.unpack(buffer[:2])[0]
 
 
-def _decode_uint_small(buffer):
-    # type: (memoryview) -> Tuple[memoryview, int]
+def _decode_uint_small(buffer: memoryview) -> Tuple[memoryview, int]:
     return buffer[1:], buffer[0]
 
 
-def _decode_uint_large(buffer):
-    # type: (memoryview) -> Tuple[memoryview, int]
+def _decode_uint_large(buffer: memoryview) -> Tuple[memoryview, int]:
     return buffer[4:], c_unsigned_int.unpack(buffer[:4])[0]
 
 
-def _decode_ulong_small(buffer):
-    # type: (memoryview) -> Tuple[memoryview, int]
+def _decode_ulong_small(buffer: memoryview)-> Tuple[memoryview, int]:
     return buffer[1:], buffer[0]
 
 
-def _decode_ulong_large(buffer):
-    # type: (memoryview) -> Tuple[memoryview, int]
+def _decode_ulong_large(buffer: memoryview)-> Tuple[memoryview, int]:
     return buffer[8:], c_unsigned_long_long.unpack(buffer[:8])[0]
 
 
-def _decode_byte(buffer):
-    # type: (memoryview) -> Tuple[memoryview, int]
+def _decode_byte(buffer: memoryview)-> Tuple[memoryview, int]:
     return buffer[1:], c_signed_char.unpack(buffer[:1])[0]
 
 
-def _decode_short(buffer):
-    # type: (memoryview) -> Tuple[memoryview, int]
+def _decode_short(buffer: memoryview)-> Tuple[memoryview, int]:
     return buffer[2:], c_signed_short.unpack(buffer[:2])[0]
 
 
-def _decode_int_small(buffer):
-    # type: (memoryview) -> Tuple[memoryview, int]
+def _decode_int_small(buffer: memoryview)-> Tuple[memoryview, int]:
     return buffer[1:], c_signed_char.unpack(buffer[:1])[0]
 
 
-def _decode_int_large(buffer):
-    # type: (memoryview) -> Tuple[memoryview, int]
+def _decode_int_large(buffer: memoryview)-> Tuple[memoryview, int]:
     return buffer[4:], c_signed_int.unpack(buffer[:4])[0]
 
 
-def _decode_long_small(buffer):
-    # type: (memoryview) -> Tuple[memoryview, int]
+def _decode_long_small(buffer: memoryview)-> Tuple[memoryview, int]:
     return buffer[1:], c_signed_char.unpack(buffer[:1])[0]
 
 
-def _decode_long_large(buffer):
-    # type: (memoryview) -> Tuple[memoryview, int]
+def _decode_long_large(buffer: memoryview)-> Tuple[memoryview, int]:
     return buffer[8:], c_signed_long_long.unpack(buffer[:8])[0]
 
 
-def _decode_float(buffer):
-    # type: (memoryview) -> Tuple[memoryview, float]
+def _decode_float(buffer: memoryview) -> Tuple[memoryview, float]:
     return buffer[4:], c_float.unpack(buffer[:4])[0]
 
 
-def _decode_double(buffer):
-    # type: (memoryview) -> Tuple[memoryview, float]
+def _decode_double(buffer: memoryview) -> Tuple[memoryview, float]:
     return buffer[8:], c_double.unpack(buffer[:8])[0]
 
 
-def _decode_timestamp(buffer):
-    # type: (memoryview) -> Tuple[memoryview, int]
+def _decode_timestamp(buffer: memoryview) -> Tuple[memoryview, int]:
     return buffer[8:], c_signed_long_long.unpack(buffer[:8])[0]
 
 
-def _decode_uuid(buffer):
-    # type: (memoryview) -> Tuple[memoryview, uuid.UUID]
+def _decode_uuid(buffer: memoryview) -> Tuple[memoryview, uuid.UUID]:
     return buffer[16:], uuid.UUID(bytes=buffer[:16].tobytes())
 
 
-def _decode_binary_small(buffer):
-    # type: (memoryview) -> Tuple[memoryview, bytes]
+def _decode_binary_small(buffer: memoryview) -> Tuple[memoryview, bytes]:
     length_index = buffer[0] + 1
     return buffer[length_index:], buffer[1:length_index].tobytes()
 
 
-def _decode_binary_large(buffer):
-    # type: (memoryview) -> Tuple[memoryview, bytes]
+def _decode_binary_large(buffer: memoryview) -> Tuple[memoryview, bytes]:
     length_index = c_unsigned_long.unpack(buffer[:4])[0] + 4
     return buffer[length_index:], buffer[4:length_index].tobytes()
 
 
-def _decode_list_small(buffer):
-    # type: (memoryview) -> Tuple[memoryview, List[Any]]
+def _decode_list_small(buffer: memoryview)-> Tuple[memoryview, List[Any]]:
     count = buffer[1]
     buffer = buffer[2:]
     values = [None] * count
@@ -171,8 +158,7 @@ def _decode_list_small(buffer):
     return buffer, values
 
 
-def _decode_list_large(buffer):
-    # type: (memoryview) -> Tuple[memoryview, List[Any]]
+def _decode_list_large(buffer: memoryview)-> Tuple[memoryview, List[Any]]:
     count = c_unsigned_long.unpack(buffer[4:8])[0]
     buffer = buffer[8:]
     values = [None] * count
@@ -181,8 +167,7 @@ def _decode_list_large(buffer):
     return buffer, values
 
 
-def _decode_map_small(buffer):
-    # type: (memoryview) -> Tuple[memoryview, Dict[Any, Any]]
+def _decode_map_small(buffer: memoryview) -> Tuple[memoryview, Dict[Any, Any]]:
     count = int(buffer[1]/2)
     buffer = buffer[2:]
     values = {}
@@ -193,8 +178,7 @@ def _decode_map_small(buffer):
     return buffer, values
 
 
-def _decode_map_large(buffer):
-    # type: (memoryview) -> Tuple[memoryview, Dict[Any, Any]]
+def _decode_map_large(buffer: memoryview) -> Tuple[memoryview, Dict[Any, Any]]:
     count = int(c_unsigned_long.unpack(buffer[4:8])[0]/2)
     buffer = buffer[8:]
     values = {}
@@ -205,8 +189,7 @@ def _decode_map_large(buffer):
     return buffer, values
 
 
-def _decode_array_small(buffer):
-    # type: (memoryview) -> Tuple[memoryview, List[Any]]
+def _decode_array_small(buffer: memoryview) -> Tuple[memoryview, List[Any]]:
     count = buffer[1]  # Ignore first byte (size) and just rely on count
     if count:
         subconstructor = buffer[2]
@@ -218,8 +201,7 @@ def _decode_array_small(buffer):
     return buffer[2:], []
 
 
-def _decode_array_large(buffer):
-    # type: (memoryview) -> Tuple[memoryview, List[Any]]
+def _decode_array_large(buffer: memoryview) -> Tuple[memoryview, List[Any]]:
     count = c_unsigned_long.unpack(buffer[4:8])[0]
     if count:
         subconstructor = buffer[8]
@@ -231,8 +213,7 @@ def _decode_array_large(buffer):
     return buffer[8:], []
 
 
-def _decode_described(buffer):
-    # type: (memoryview) -> Tuple[memoryview, Any]
+def _decode_described(buffer: memoryview) -> Tuple[memoryview, object]:
     # TODO: to move the cursor of the buffer to the described value based on size of the
     #  descriptor without decoding descriptor value
     composite_type = buffer[0]
@@ -245,8 +226,7 @@ def _decode_described(buffer):
         return buffer, value
 
 
-def decode_payload(buffer):
-    # type: (memoryview) -> Message
+def decode_payload(buffer: memoryview) -> Message:
     message: Dict[str, Union[Properties, Header, Dict, bytes, List]] = {}
     while buffer:
         # Ignore the first two bytes, they will always be the constructors for
@@ -279,12 +259,12 @@ def decode_payload(buffer):
             message["footer"] = value
     # TODO: we can possibly swap out the Message construct with a TypedDict
     #  for both input and output so we get the best of both.
+    # casting to TypedDict with named fields to allow for unpacking with **
     message_properties = cast("MessageDict", message)
     return Message(**message_properties)
 
 
-def decode_frame(data):
-    # type: (memoryview) -> Tuple[int, List[Any]]
+def decode_frame(data: memoryview) -> Tuple[int, List[Any]]:
     # Ignore the first two bytes, they will always be the constructors for
     # described type then ulong.
     frame_type = data[2]
@@ -305,8 +285,7 @@ def decode_frame(data):
     return frame_type, fields
 
 
-def decode_empty_frame(header):
-    # type: (memoryview) -> Tuple[int, bytes]
+def decode_empty_frame(header: memoryview) -> Tuple[int, bytes]:
     if header[0:4] == _HEADER_PREFIX:
         return 0, header.tobytes()
     if header[5] == 0: