[pyamqp] pylint/mypy/black

tests/test_decode.py

@@ -8,7 +8,7 @@
 import uuid
 
 from uamqp import _decode as decode
-from uamqp.types import AMQPTypes, TYPE, VALUE
+from uamqp.amqp_types import AMQPTypes, TYPE, VALUE
 
 import pytest
 

tests/test_encode.py

@@ -8,7 +8,7 @@
 import uuid
 
 import uamqp._encode as encode
-from uamqp.types import AMQPTypes, TYPE, VALUE
+from uamqp.amqp_types import AMQPTypes, TYPE, VALUE
 from uamqp.message import Message, Header, Properties
 
 import pytest

uamqp/__init__.py

@@ -1,13 +1,13 @@
-#-------------------------------------------------------------------------
+# -------------------------------------------------------------------------
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
-#-------------------------------------------------------------------------
+# -------------------------------------------------------------------------
 
 __version__ = "2.0.0a1"
 
 
-from ._connection import Connection
-from ._transport import SSLTransport
+from uamqp._connection import Connection
+from uamqp._transport import SSLTransport
 
-from .client import AMQPClient, ReceiveClient, SendClient
+from uamqp.client import AMQPClient, ReceiveClient, SendClient

uamqp/_decode.py

@@ -1,39 +1,46 @@
-#-------------------------------------------------------------------------
+# -------------------------------------------------------------------------
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
-#--------------------------------------------------------------------------
+# --------------------------------------------------------------------------
 # pylint: disable=redefined-builtin, import-error
 
+import logging
 import struct
 import uuid
-import logging
-from typing import List, Union, Tuple, Dict, Callable  # pylint: disable=unused-import
-
-
-from .message import Message, Header, Properties
+from typing import (
+    List,
+    Tuple,
+    Dict,
+    Callable,
+    Any,
+    Union,
+    Optional,
+)  # pylint: disable=unused-import
+
+from uamqp.message import Message, Header, Properties
 
 _LOGGER = logging.getLogger(__name__)
-_HEADER_PREFIX = memoryview(b'AMQP')
+_HEADER_PREFIX = memoryview(b"AMQP")
 _COMPOSITES = {
-    35: 'received',
-    36: 'accepted',
-    37: 'rejected',
-    38: 'released',
-    39: 'modified',
+    35: "received",
+    36: "accepted",
+    37: "rejected",
+    38: "released",
+    39: "modified",
 }
 
-c_unsigned_char = struct.Struct('>B')
-c_signed_char = struct.Struct('>b')
-c_unsigned_short = struct.Struct('>H')
-c_signed_short = struct.Struct('>h')
-c_unsigned_int = struct.Struct('>I')
-c_signed_int = struct.Struct('>i')
-c_unsigned_long = struct.Struct('>L')
-c_unsigned_long_long = struct.Struct('>Q')
-c_signed_long_long = struct.Struct('>q')
-c_float = struct.Struct('>f')
-c_double = struct.Struct('>d')
+c_unsigned_char = struct.Struct(">B")
+c_signed_char = struct.Struct(">b")
+c_unsigned_short = struct.Struct(">H")
+c_signed_short = struct.Struct(">h")
+c_unsigned_int = struct.Struct(">I")
+c_signed_int = struct.Struct(">i")
+c_unsigned_long = struct.Struct(">L")
+c_unsigned_long_long = struct.Struct(">Q")
+c_signed_long_long = struct.Struct(">q")
+c_float = struct.Struct(">f")
+c_double = struct.Struct(">d")
 
 
 def _decode_null(buffer):
@@ -63,7 +70,7 @@ def _decode_empty(buffer):
 
 def _decode_boolean(buffer):
     # type: (memoryview) -> Tuple[memoryview, bool]
-    return buffer[1:], buffer[:1] == b'\x01'
+    return buffer[1:], buffer[:1] == b"\x01"
 
 
 def _decode_ubyte(buffer):
@@ -164,7 +171,7 @@ def _decode_list_small(buffer):
     buffer = buffer[2:]
     values = [None] * count
     for i in range(count):
-        buffer, values[i] = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:])
+        buffer, values[i] = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:])  # type: ignore
     return buffer, values
 
 
@@ -174,30 +181,30 @@ def _decode_list_large(buffer):
     buffer = buffer[8:]
     values = [None] * count
     for i in range(count):
-        buffer, values[i] = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:])
+        buffer, values[i] = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:])  # type: ignore
     return buffer, values
 
 
 def _decode_map_small(buffer):
     # type: (memoryview) -> Tuple[memoryview, Dict[Any, Any]]
-    count = int(buffer[1]/2)
+    count = int(buffer[1] / 2)
     buffer = buffer[2:]
     values = {}
-    for  _ in range(count):
-        buffer, key = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:])
-        buffer, value = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:])
+    for _ in range(count):
+        buffer, key = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:])  # type: ignore
+        buffer, value = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:])  # type: ignore
         values[key] = value
     return buffer, values
 
 
 def _decode_map_large(buffer):
     # type: (memoryview) -> Tuple[memoryview, Dict[Any, Any]]
-    count = int(c_unsigned_long.unpack(buffer[4:8])[0]/2)
+    count = int(c_unsigned_long.unpack(buffer[4:8])[0] / 2)
     buffer = buffer[8:]
     values = {}
-    for  _ in range(count):
-        buffer, key = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:])
-        buffer, value = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:])
+    for _ in range(count):
+        buffer, key = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:])  # type: ignore
+        buffer, value = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:])  # type: ignore
         values[key] = value
     return buffer, values
 
@@ -210,7 +217,7 @@ def _decode_array_small(buffer):
         buffer = buffer[3:]
         values = [None] * count
         for i in range(count):
-            buffer, values[i] = _DECODE_BY_CONSTRUCTOR[subconstructor](buffer)
+            buffer, values[i] = _DECODE_BY_CONSTRUCTOR[subconstructor](buffer)  # type: ignore
         return buffer, values
     return buffer[2:], []
 
@@ -223,7 +230,7 @@ def _decode_array_large(buffer):
         buffer = buffer[9:]
         values = [None] * count
         for i in range(count):
-            buffer, values[i] = _DECODE_BY_CONSTRUCTOR[subconstructor](buffer)
+            buffer, values[i] = _DECODE_BY_CONSTRUCTOR[subconstructor](buffer)  # type: ignore
         return buffer, values
     return buffer[8:], []
 
@@ -233,23 +240,23 @@ def _decode_described(buffer):
     # 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]
-    buffer, descriptor = _DECODE_BY_CONSTRUCTOR[composite_type](buffer[1:])
-    buffer, value = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:])
+    buffer, descriptor = _DECODE_BY_CONSTRUCTOR[composite_type](buffer[1:])  # type: ignore
+    buffer, value = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:])  # type: ignore
     try:
-        composite_type = _COMPOSITES[descriptor]
+        composite_type = _COMPOSITES[descriptor]  # type: ignore
         return buffer, {composite_type: value}
     except KeyError:
         return buffer, value
 
 
 def decode_payload(buffer):
     # type: (memoryview) -> Message
-    message = {}
+    message: Dict[str, Any] = {}
     while buffer:
         # Ignore the first two bytes, they will always be the constructors for
         # described type then ulong.
         descriptor = buffer[2]
-        buffer, value = _DECODE_BY_CONSTRUCTOR[buffer[3]](buffer[4:])
+        buffer, value = _DECODE_BY_CONSTRUCTOR[buffer[3]](buffer[4:])  # type: ignore
         if descriptor == 112:
             message["header"] = Header(*value)
         elif descriptor == 113:
@@ -285,32 +292,34 @@ def decode_frame(data):
     # described type then ulong.
     frame_type = data[2]
     compound_list_type = data[3]
-    if compound_list_type == 0xd0:
+    if compound_list_type == 0xD0:
         # list32 0xd0: data[4:8] is size, data[8:12] is count
         count = c_signed_int.unpack(data[8:12])[0]
         buffer = data[12:]
     else:
         # list8 0xc0: data[4] is size, data[5] is count
         count = data[5]
         buffer = data[6:]
-    fields = [None] * count
+    fields: List[Union[None, memoryview]] = [None] * count
     for i in range(count):
-        buffer, fields[i] = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:])
+        buffer, fields[i] = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:])  # type: ignore
     if frame_type == 20:
         fields.append(buffer)
     return frame_type, fields
 
 
 def decode_empty_frame(header):
-    # type: (memory) -> bytes
+    # type: (memoryview) -> Tuple[int,bytes]
     if header[0:4] == _HEADER_PREFIX:
         return 0, header.tobytes()
     if header[5] == 0:
         return 1, b"EMPTY"
     raise ValueError("Received unrecognized empty frame")
 
 
-_DECODE_BY_CONSTRUCTOR = [None] * 256  # type: List[Callable[memoryview]]
+_DECODE_BY_CONSTRUCTOR = [
+    None
+] * 256  # type: List[Optional[Callable[[memoryview], Tuple[memoryview, Any]]]]
 _DECODE_BY_CONSTRUCTOR[0] = _decode_described
 _DECODE_BY_CONSTRUCTOR[64] = _decode_null
 _DECODE_BY_CONSTRUCTOR[65] = _decode_true