mypy fixes

Author: mattsb42-aws

src/dynamodb_encryption_sdk/delegated_keys/__init__.py

@@ -13,7 +13,7 @@
 """Delegated keys."""
 import abc
 try:  # Python 3.5.0 and 3.5.1 have incompatible typing modules
-    from typing import Dict, Text  # noqa pylint: disable=unused-import
+    from typing import Dict, Optional, Text  # noqa pylint: disable=unused-import
 except ImportError:  # pragma: no cover
     # We only actually need these imports when running the mypy checks
     pass
@@ -43,17 +43,24 @@ class DelegatedKey(object):
     a ``NotImplementedError`` detailing this.
     """
 
-    #: Most delegated keys should not be used with RawCryptographicMaterials.
-    allowed_for_raw_materials = False
-
     @abc.abstractproperty
     def algorithm(self):
         # type: () -> Text
         """Text description of algorithm used by this delegated key."""
 
+    @property
+    def allowed_for_raw_materials(self):
+        # type: () -> bool
+        """Most delegated keys should not be used with RawCryptographicMaterials.
+
+        :returns: False
+        :rtype: bool
+        """
+        return False
+
     @classmethod
-    def generate(cls, algorithm, key_length):
-        # type: (Text, int) -> None
+    def generate(cls, algorithm, key_length):  # type: ignore
+        # type: (Text, int) -> DelegatedKey
         # pylint: disable=unused-argument,no-self-use
         """Generate an instance of this DelegatedKey using the specified algorithm and key length.
 
@@ -64,8 +71,8 @@ def generate(cls, algorithm, key_length):
         """
         _raise_not_implemented('generate')
 
-    def encrypt(self, algorithm, name, plaintext, additional_associated_data=None):
-        # type: (Text, Text, bytes, Dict[Text, Text]) -> bytes
+    def encrypt(self, algorithm, name, plaintext, additional_associated_data=None):  # type: ignore
+        # type: (Text, Text, bytes, Optional[Dict[Text, Text]]) -> bytes
         # pylint: disable=unused-argument,no-self-use
         """Encrypt data.
 
@@ -79,8 +86,8 @@ def encrypt(self, algorithm, name, plaintext, additional_associated_data=None):
         """
         _raise_not_implemented('encrypt')
 
-    def decrypt(self, algorithm, name, ciphertext, additional_associated_data=None):
-        # type: (Text, Text, bytes, Dict[Text, Text]) -> bytes
+    def decrypt(self, algorithm, name, ciphertext, additional_associated_data=None):  # type: ignore
+        # type: (Text, Text, bytes, Optional[Dict[Text, Text]]) -> bytes
         # pylint: disable=unused-argument,no-self-use
         """Encrypt data.
 
@@ -94,8 +101,8 @@ def decrypt(self, algorithm, name, ciphertext, additional_associated_data=None):
         """
         _raise_not_implemented('decrypt')
 
-    def wrap(self, algorithm, content_key, additional_associated_data=None):
-        # type: (Text, bytes, Dict[Text, Text]) -> bytes
+    def wrap(self, algorithm, content_key, additional_associated_data=None):  # type: ignore
+        # type: (Text, bytes, Optional[Dict[Text, Text]]) -> bytes
         # pylint: disable=unused-argument,no-self-use
         """Wrap content key.
 
@@ -108,8 +115,15 @@ def wrap(self, algorithm, content_key, additional_associated_data=None):
         """
         _raise_not_implemented('wrap')
 
-    def unwrap(self, algorithm, wrapped_key, wrapped_key_algorithm, wrapped_key_type, additional_associated_data=None):
-        # type: (Text, bytes, Text, EncryptionKeyType, Dict[Text, Text]) -> DelegatedKey
+    def unwrap(  # type: ignore
+            self,
+            algorithm,
+            wrapped_key,
+            wrapped_key_algorithm,
+            wrapped_key_type,
+            additional_associated_data=None
+    ):
+        # type: (Text, bytes, Text, EncryptionKeyType, Optional[Dict[Text, Text]]) -> DelegatedKey
         # pylint: disable=unused-argument,no-self-use
         """Wrap content key.
 
@@ -125,7 +139,7 @@ def unwrap(self, algorithm, wrapped_key, wrapped_key_algorithm, wrapped_key_type
         """
         _raise_not_implemented('unwrap')
 
-    def sign(self, algorithm, data):
+    def sign(self, algorithm, data):  # type: ignore
         # type: (Text, bytes) -> bytes
         # pylint: disable=unused-argument,no-self-use
         """Sign data.
@@ -137,7 +151,7 @@ def sign(self, algorithm, data):
         """
         _raise_not_implemented('sign')
 
-    def verify(self, algorithm, signature, data):
+    def verify(self, algorithm, signature, data):  # type: ignore
         # type: (Text, bytes, bytes) -> None
         # pylint: disable=unused-argument,no-self-use
         """Sign data.
@@ -148,10 +162,10 @@ def verify(self, algorithm, signature, data):
         """
         _raise_not_implemented('verify')
 
-    def signing_algorithm(self):
+    def signing_algorithm(self):  # type: ignore
         # type: () -> Text
         # pylint: disable=no-self-use
-        """Provides a description that can inform an appropriate cryptographic materials
+        """Provide a description that can inform an appropriate cryptographic materials
         provider about how to build a DelegatedKey for signature verification. If implemented,
         the return value of this method is included in the material description written to
         the encrypted item.

src/dynamodb_encryption_sdk/delegated_keys/jce.py

@@ -228,6 +228,7 @@ def generate(cls, algorithm, key_length=None):
 
     @property
     def allowed_for_raw_materials(self):
+        # type: () -> bool
         """Only ``JceNameLocalDelegatedKey`` backed by AES keys are allowed to be used with
         ``RawCryptographicMaterials``.
 
@@ -237,7 +238,7 @@ def allowed_for_raw_materials(self):
         return self.algorithm == 'AES'
 
     def _encrypt(self, algorithm, name, plaintext, additional_associated_data=None):
-        # type: (Text, Text, bytes, Dict[Text, Text]) -> bytes
+        # type: (Text, Text, bytes, Optional[Dict[Text, Text]]) -> bytes
         # pylint: disable=unused-argument
         """
         Encrypt data.
@@ -255,7 +256,7 @@ def _encrypt(self, algorithm, name, plaintext, additional_associated_data=None):
         return encryptor.encrypt(self.__key, plaintext)
 
     def _decrypt(self, algorithm, name, ciphertext, additional_associated_data=None):
-        # type: (Text, Text, bytes, Dict[Text, Text]) -> bytes
+        # type: (Text, Text, bytes, Optional[Dict[Text, Text]]) -> bytes
         # pylint: disable=unused-argument
         """Encrypt data.
 
@@ -271,7 +272,7 @@ def _decrypt(self, algorithm, name, ciphertext, additional_associated_data=None)
         return decryptor.decrypt(self.__key, ciphertext)
 
     def _wrap(self, algorithm, content_key, additional_associated_data=None):
-        # type: (Text, bytes, Dict[Text, Text]) -> bytes
+        # type: (Text, bytes, Optional[Dict[Text, Text]]) -> bytes
         # pylint: disable=unused-argument
         """Wrap content key.
 
@@ -288,7 +289,7 @@ def _wrap(self, algorithm, content_key, additional_associated_data=None):
         )
 
     def _unwrap(self, algorithm, wrapped_key, wrapped_key_algorithm, wrapped_key_type, additional_associated_data=None):
-        # type: (Text, bytes, Text, EncryptionKeyType, Dict[Text, Text]) -> DelegatedKey
+        # type: (Text, bytes, Text, EncryptionKeyType, Optional[Dict[Text, Text]]) -> DelegatedKey
         # pylint: disable=unused-argument
         """Wrap content key.
 
@@ -341,7 +342,7 @@ def _verify(self, algorithm, signature, data):
 
     def _signing_algorithm(self):
         # type: () -> Text
-        """Provides a description that can inform an appropriate cryptographic materials
+        """Provide a description that can inform an appropriate cryptographic materials
         provider about how to build a ``JceNameLocalDelegatedKey`` for signature verification.
         The return value of this method is included in the material description written to
         the encrypted item.

src/dynamodb_encryption_sdk/encrypted/client.py

@@ -17,7 +17,7 @@
 import botocore
 
 try:  # Python 3.5.0 and 3.5.1 have incompatible typing modules
-    from typing import Any, Callable, Dict, Optional  # noqa pylint: disable=unused-import
+    from typing import Any, Callable, Dict, Iterator, Optional  # noqa pylint: disable=unused-import
 except ImportError:  # pragma: no cover
     # We only actually need these imports when running the mypy checks
     pass
@@ -88,7 +88,7 @@ def __getattr__(self, name):
         return getattr(self._paginator, name)
 
     def paginate(self, **kwargs):
-        # type: (**Any) -> Dict
+        # type: (**Any) -> Iterator[Dict]
         # TODO: narrow this down
         """Create an iterator that will paginate through responses from the underlying paginator,
         transparently decrypting any returned items.

src/dynamodb_encryption_sdk/encrypted/resource.py

@@ -17,6 +17,12 @@
 from boto3.resources.base import ServiceResource
 from boto3.resources.collection import CollectionManager
 
+try:  # Python 3.5.0 and 3.5.1 have incompatible typing modules
+    from typing import Optional  # noqa pylint: disable=unused-import
+except ImportError:  # pragma: no cover
+    # We only actually need these imports when running the mypy checks
+    pass
+
 from dynamodb_encryption_sdk.internal.utils import (
     crypto_config_from_cache, decrypt_batch_get_item, encrypt_batch_write_item, TableInfoCache
 )

src/dynamodb_encryption_sdk/encrypted/table.py

@@ -17,6 +17,12 @@
 from boto3.dynamodb.table import BatchWriter
 from boto3.resources.base import ServiceResource
 
+try:  # Python 3.5.0 and 3.5.1 have incompatible typing modules
+    from typing import Optional  # noqa pylint: disable=unused-import
+except ImportError:  # pragma: no cover
+    # We only actually need these imports when running the mypy checks
+    pass
+
 from dynamodb_encryption_sdk.internal.utils import (
     crypto_config_from_kwargs, crypto_config_from_table_info,
     decrypt_get_item, decrypt_multi_get, encrypt_put_item

src/dynamodb_encryption_sdk/internal/crypto/jce_bridge/primitives.py

@@ -23,7 +23,7 @@
 import six
 
 try:  # Python 3.5.0 and 3.5.1 have incompatible typing modules
-    from typing import Callable, Text  # noqa pylint: disable=unused-import
+    from typing import Any, Callable, Text  # noqa pylint: disable=unused-import
 except ImportError:  # pragma: no cover
     # We only actually need these imports when running the mypy checks
     pass
@@ -268,8 +268,10 @@ def __init__(
         self.java_name = java_name
         self.cipher = cipher
         attr.validate(self)
-        if hasattr(self, '__attrs_post_init__'):
-            self.__attrs_post_init__()
+        self.__attrs_post_init__()
+
+    def __attrs_post_init__(self):
+        """No-op stub to standardize API."""
 
     def validate_algorithm(self, algorithm):
         # type: (Text) -> None

src/dynamodb_encryption_sdk/internal/dynamodb_types.py

@@ -1,14 +1,14 @@
 """Types used with mypy for DynamoDB items and attributes."""
 try:  # Python 3.5.0 and 3.5.1 have incompatible typing modules
-    from typing import Any, ByteString, Dict, List, Text, Union
+    from typing import Any, AnyStr, ByteString, Dict, List, Text
 
-    ATTRIBUTE = Any  # TODO: narrow this down
+    ATTRIBUTE = Dict[Text, Any]  # TODO: narrow this down
     ITEM = Dict[Text, ATTRIBUTE]
     RAW_ATTRIBUTE = ITEM
     NULL = bool  # DynamoDB TypeSerializer converts none to {'NULL': True}
     BOOLEAN = bool
     NUMBER = int  # TODO: This misses long on Python 2...figure out something for this
-    STRING = Union[Text, Text]  # TODO: can be unicode but should not be bytes
+    STRING = AnyStr  # TODO: can be unicode but should not be bytes
     BINARY = ByteString
     BINARY_ATTRIBUTE = Dict[Text, BINARY]
     SET = List  # DynamoDB TypeSerializer converts sets into lists

src/dynamodb_encryption_sdk/internal/formatting/deserialize/attribute.py

@@ -18,7 +18,7 @@
 import struct
 
 try:  # Python 3.5.0 and 3.5.1 have incompatible typing modules
-    from typing import Callable, Dict, List, Union  # noqa pylint: disable=unused-import
+    from typing import Callable, Dict, List, Text, Union  # noqa pylint: disable=unused-import
     from dynamodb_encryption_sdk.internal import dynamodb_types  # noqa pylint: disable=unused-import,ungrouped-imports
 except ImportError:  # pragma: no cover
     # We only actually need these imports when running the mypy checks
@@ -52,7 +52,7 @@ def _transform_binary_value(value):
         return value
 
     def _deserialize_binary(stream):
-        # type: (io.BytesIO) -> Dict[str, bytes]
+        # type: (io.BytesIO) -> Dict[Text, bytes]
         """Deserializes a binary object.
 
         :param stream: Stream containing serialized object
@@ -72,7 +72,7 @@ def _transform_string_value(value):
         return codecs.decode(value, TEXT_ENCODING)
 
     def _deserialize_string(stream):
-        # type: (io.BytesIO) -> Dict[str, dynamodb_types.STRING]
+        # type: (io.BytesIO) -> Dict[Text, dynamodb_types.STRING]
         """Deserializes a string object.
 
         :param stream: Stream containing serialized object
@@ -94,7 +94,7 @@ def _transform_number_value(value):
         return '{0:f}'.format(decimal_value)
 
     def _deserialize_number(stream):
-        # type: (io.BytesIO) -> Dict[str, dynamodb_types.STRING]
+        # type: (io.BytesIO) -> Dict[Text, dynamodb_types.STRING]
         """Deserializes a number object.
 
         :param stream: Stream containing serialized object
@@ -110,7 +110,7 @@ def _deserialize_number(stream):
     }
 
     def _deserialize_boolean(stream):
-        # type: (io.BytesIO) -> Dict[str, dynamodb_types.BOOLEAN]
+        # type: (io.BytesIO) -> Dict[Text, dynamodb_types.BOOLEAN]
         """Deserializes a boolean object.
 
         :param stream: Stream containing serialized object
@@ -121,7 +121,7 @@ def _deserialize_boolean(stream):
         return {Tag.BOOLEAN.dynamodb_tag: _boolean_map[value]}
 
     def _deserialize_null(stream):  # we want a consistent API but don't use stream, so pylint: disable=unused-argument
-        # type: (io.BytesIO) -> Dict[str, dynamodb_types.BOOLEAN]
+        # type: (io.BytesIO) -> Dict[Text, dynamodb_types.BOOLEAN]
         """Deserializes a null object.
 
         :param stream: Stream containing serialized object
@@ -145,7 +145,7 @@ def _deserialize_set(stream, member_transform):
         ])
 
     def _deserialize_binary_set(stream):
-        # type: (io.BytesIO) -> Dict[str, dynamodb_types.SET[dynamodb_types.BINARY]]
+        # type: (io.BytesIO) -> Dict[Text, dynamodb_types.SET[dynamodb_types.BINARY]]
         """Deserializes a binary set object.
 
         :param stream: Stream containing serialized object
@@ -155,7 +155,7 @@ def _deserialize_binary_set(stream):
         return {Tag.BINARY_SET.dynamodb_tag: _deserialize_set(stream, _transform_binary_value)}
 
     def _deserialize_string_set(stream):
-        # type: (io.BytesIO) -> Dict[str, dynamodb_types.SET[dynamodb_types.STRING]]
+        # type: (io.BytesIO) -> Dict[Text, dynamodb_types.SET[dynamodb_types.STRING]]
         """Deserializes a string set object.
 
         :param stream: Stream containing serialized object
@@ -165,7 +165,7 @@ def _deserialize_string_set(stream):
         return {Tag.STRING_SET.dynamodb_tag: _deserialize_set(stream, _transform_string_value)}
 
     def _deserialize_number_set(stream):
-        # type: (io.BytesIO) -> Dict[str, dynamodb_types.SET[dynamodb_types.STRING]]
+        # type: (io.BytesIO) -> Dict[Text, dynamodb_types.SET[dynamodb_types.STRING]]
         """Deserializes a number set object.
 
         :param stream: Stream containing serialized object
@@ -175,7 +175,7 @@ def _deserialize_number_set(stream):
         return {Tag.NUMBER_SET.dynamodb_tag: _deserialize_set(stream, _transform_number_value)}
 
     def _deserialize_list(stream):
-        # type: (io.BytesIO) -> Dict[str, dynamodb_types.LIST]
+        # type: (io.BytesIO) -> Dict[Text, dynamodb_types.LIST]
         """Deserializes a list object.
 
         :param stream: Stream containing serialized object
@@ -189,15 +189,15 @@ def _deserialize_list(stream):
         ]}
 
     def _deserialize_map(stream):
-        # type: (io.BytesIO) -> Dict[str, dynamodb_types.MAP]
+        # type: (io.BytesIO) -> Dict[Text, dynamodb_types.MAP]
         """Deserializes a map object.
 
         :param stream: Stream containing serialized object
         :type stream: io.BytesIO
         :rtype: dict
         """
         member_count = decode_length(stream)
-        members = {}
+        members = {}  # type: dynamodb_types.MAP
         for _ in range(member_count):
             key = _deserialize(stream)
             if Tag.STRING.dynamodb_tag not in key:
@@ -236,7 +236,7 @@ def _deserialize_function(tag):
             raise DeserializationError('Unsupported tag: "{}"'.format(tag))
 
     def _deserialize(stream):
-        # type: (io.BytesIO) -> Dict[str, dynamodb_types.RAW_ATTRIBUTE]
+        # type: (io.BytesIO) -> Dict[Text, dynamodb_types.RAW_ATTRIBUTE]
         """Deserializes a serialized object.
 
         :param stream: Stream containing serialized object