Add mypy check to github actions

.github/workflows/python-package.yml

@@ -32,6 +32,10 @@ jobs:
     - name: Test with pytest
       run: |
         python setup.py test
+    - name: Run mypy
+      continue-on-error: true
+      run: |
+        mypy pyms
     - name: Lint with flake8
       run: |
         # stop the build if there are Python syntax errors or undefined names

pyms/cloud/aws/kms.py

@@ -8,36 +8,36 @@ class Crypt(CryptAbstract):
     encryption_algorithm = "SYMMETRIC_DEFAULT"  # 'SYMMETRIC_DEFAULT' | 'RSAES_OAEP_SHA_1' | 'RSAES_OAEP_SHA_256'
     key_id = ""
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args, **kwargs) -> None:
         self._init_boto()
         super().__init__(*args, **kwargs)
 
-    def encrypt(self, message):   # pragma: no cover
+    def encrypt(self, message: str) -> str:   # pragma: no cover
         ciphertext = self.client.encrypt(
             KeyId=self.config.key_id,
             Plaintext=bytes(message, encoding="UTF-8"),
         )
         return str(base64.b64encode(ciphertext["CiphertextBlob"]), encoding="UTF-8")
 
-    def _init_boto(self):  # pragma: no cover
+    def _init_boto(self) -> None:  # pragma: no cover
         check_package_exists("boto3")
         boto3 = import_package("boto3")
         boto3.set_stream_logger(name='botocore')
         self.client = boto3.client('kms')
 
-    def _aws_decrypt(self, blob_text):  # pragma: no cover
+    def _aws_decrypt(self, blob_text: bytes) -> str:  # pragma: no cover
         response = self.client.decrypt(
             CiphertextBlob=blob_text,
             KeyId=self.config.key_id,
             EncryptionAlgorithm=self.encryption_algorithm
         )
         return str(response['Plaintext'], encoding="UTF-8")
 
-    def _parse_encrypted(self, encrypted):
+    def _parse_encrypted(self, encrypted: str) -> bytes:
         blob_text = base64.b64decode(encrypted)
         return blob_text
 
-    def decrypt(self, encrypted):
+    def decrypt(self, encrypted: str) -> str:
         blob_text = self._parse_encrypted(encrypted)
         decrypted = self._aws_decrypt(blob_text)
 

pyms/crypt/fernet.py

@@ -18,8 +18,8 @@ def __init__(self, *args, **kwargs):
         self._loader = LoadFile(kwargs.get("path"), CRYPT_FILE_KEY_ENVIRONMENT, DEFAULT_KEY_FILENAME)
         super().__init__(*args, **kwargs)
 
-    def generate_key(self, password: Text, write_to_file: bool = False):
-        password = password.encode()  # Convert to type bytes
+    def generate_key(self, password: Text, write_to_file: bool = False) -> bytes:
+        byte_password = password.encode()  # Convert to type bytes
         salt = os.urandom(16)
         kdf = PBKDF2HMAC(
             algorithm=hashes.SHA512_256(),
@@ -28,7 +28,7 @@ def generate_key(self, password: Text, write_to_file: bool = False):
             iterations=100000,
             backend=default_backend()
         )
-        key = base64.urlsafe_b64encode(kdf.derive(password))  # Can only use kdf once
+        key = base64.urlsafe_b64encode(kdf.derive(byte_password))  # Can only use kdf once
         if write_to_file:
             self._loader.put_file(key, 'wb')
         return key

pyms/flask/app/create_app.py

@@ -1,6 +1,6 @@
 import logging
 import os
-from typing import List
+from typing import List, Optional
 
 from flask import Flask
 
@@ -73,12 +73,12 @@ def example():
     Current services are swagger, request, tracer, metrics
     """
     config_resource = CONFIG_BASE
-    services: List[DriverService] = []
+    services: List[str] = []
     application = Flask
-    swagger: DriverService = None
-    request: DriverService = None
-    tracer: DriverService = None
-    metrics: DriverService = None
+    swagger: Optional[DriverService] = None
+    request: Optional[DriverService] = None
+    tracer: Optional[DriverService] = None
+    metrics: Optional[DriverService] = None
     _singleton = True
 
     def __init__(self, *args, **kwargs):

pyms/flask/services/driver.py

@@ -9,7 +9,7 @@
 logger = logging.getLogger(LOGGER_NAME)
 
 
-def get_service_name(service_base=SERVICE_BASE, service=""):
+def get_service_name(service_base: str = SERVICE_BASE, service: str = "") -> str:
     return ".".join([service_base, service])
 
 
@@ -48,10 +48,10 @@ def __getattr__(self, attr, *args, **kwargs):
         return config_attribute if config_attribute == "" or config_attribute != {} else self.default_values.get(attr,
                                                                                                                  None)
 
-    def is_enabled(self):
+    def is_enabled(self) -> bool:
         return self.enabled
 
-    def exists_config(self):
+    def exists_config(self) -> bool:
         return self.config is not None and isinstance(self.config, ConfFile)
 
 

pyms/flask/services/requests.py

@@ -97,7 +97,7 @@ def set_propagate_headers(headers: dict) -> dict:
                 headers.update({k: v})
         return headers
 
-    def _get_headers(self, headers, propagate_headers=False):
+    def _get_headers(self, headers: dict, propagate_headers: bool = False) -> dict:
         """If enabled appends trace headers to received ones.
 
         :param headers: dictionary of HTTP Headers to send.

setup.cfg

@@ -8,4 +8,7 @@ addopts = --cov=pyms --cov=tests tests/
 ignore = E501
 exclude = .git,__pycache__,docs/source/conf.py,old,build,dist
 max-complexity = 12
-max-line-length = 120
+max-line-length = 120
+
+[mypy]
+ignore_missing_imports = True

setup.py

@@ -64,7 +64,8 @@
     'mkdocs>=1.1.2',
     'mkdocs-material>=6.0.0',
     'lightstep>=4.4.8',
-    'safety==1.9.0'
+    'safety==1.9.0',
+    'mypy>=0.782'
 ]
 
 install_all_requires = (install_request_requires + install_swagger_requires +