Type hinting (#214)

* refactor: add type hint flask module

Author: AmandaKhol

.gitignore

@@ -16,6 +16,7 @@ py_ms.egg-info/*
 .eggs/*
 pylintReport.txt
 .scannerwork/
+.mypy_cache
 
 # Deploy
 build/

pyms/flask/app/create_app.py

@@ -205,7 +205,7 @@ def reload_conf(self):
         self.crypt.config.reload()
         self.create_app()
 
-    def create_app(self):
+    def create_app(self) -> Flask:
         """Initialize the Flask app, register blueprints and initialize
         all libraries like Swagger, database,
         the trace system...
@@ -234,13 +234,13 @@ def create_app(self):
 
         return self.application
 
-    def add_error_handlers(self):
+    def add_error_handlers(self) -> None:
         """Subclasses will override this method in order to add specific error handlers. This should be done with
         calls to add_error_handler method.
         """
         pass
 
-    def add_error_handler(self, code_or_exception, handler):
+    def add_error_handler(self, code_or_exception, handler) -> None:
         """Add custom handler for an error code or exception in the connexion app.
 
         :param code_or_exception: HTTP error code or exception

pyms/flask/app/utils.py

@@ -10,7 +10,7 @@ class SingletonMeta(type):
     _instances: Dict[type, type] = {}
     _singleton = True
 
-    def __call__(cls, *args, **kwargs):
+    def __call__(cls, *args, **kwargs) -> type:
         if cls not in cls._instances or not cls._singleton:
             cls._instances[cls] = super(SingletonMeta, cls).__call__(*args, **kwargs)
         else:
@@ -32,7 +32,7 @@ def __init__(self, app):
         self.app = app
 
     @staticmethod
-    def _extract_prefix(environ):
+    def _extract_prefix(environ: dict) -> str:
         """
         Get Path from environment from:
         - Traefik with HTTP_X_SCRIPT_NAME https://docs.traefik.io/v2.0/middlewares/headers/

pyms/flask/services/metrics.py

@@ -1,8 +1,9 @@
 import time
 import logging
+
 from typing import Text
 
-from flask import Blueprint, Response, request
+from flask import Blueprint, Response, request, Flask
 from prometheus_client import multiprocess, Counter, Histogram, generate_latest, CollectorRegistry, REGISTRY
 from pyms.flask.services.driver import DriverService
 
@@ -27,10 +28,10 @@ class FlaskMetricsWrapper():
     def __init__(self, app_name):
         self.app_name = app_name
 
-    def before_request(self):  # pylint: disable=R0201
+    def before_request(self) -> None:  # pylint: disable=R0201
         request.start_time = time.time()
 
-    def after_request(self, response):
+    def after_request(self, response: Response) -> Response:
         if hasattr(request.url_rule, "rule"):
             path = request.url_rule.rule
         else:
@@ -54,7 +55,7 @@ def __init__(self, *args, **kwargs):
         self.init_registry()
         self.serve_metrics()
 
-    def init_registry(self):
+    def init_registry(self) -> None:
         try:
             multiprocess_registry = CollectorRegistry()
             multiprocess.MultiProcessCollector(multiprocess_registry)
@@ -63,7 +64,7 @@ def init_registry(self):
             self.registry = REGISTRY
 
     @staticmethod
-    def monitor(app_name, app):
+    def monitor(app_name: str, app: Flask) -> None:
         metric = FlaskMetricsWrapper(app_name)
         app.before_request(metric.before_request)
         app.after_request(metric.after_request)
@@ -78,7 +79,7 @@ def metrics():  # pylint: disable=unused-variable
             )
 
     @staticmethod
-    def add_logger_handler(logger, service_name):
+    def add_logger_handler(logger: logging.Logger, service_name: str) -> logging.Logger:
         logger.addHandler(MetricsLogHandler(service_name))
         return logger
 
@@ -90,5 +91,5 @@ def __init__(self, app_name):
         super().__init__()
         self.app_name = app_name
 
-    def emit(self, record):
+    def emit(self, record) -> None:
         LOGGER_TOTAL_MESSAGES.labels(self.app_name, record.levelname).inc()

pyms/flask/services/requests.py

@@ -1,10 +1,11 @@
 """Encapsulate common rest operations between business services propagating trace headers if configured.
 """
 import logging
+from typing import Any
 
 import requests
 from flask import request
-from requests.adapters import HTTPAdapter
+from requests.adapters import HTTPAdapter, Response
 from urllib3.util.retry import Retry
 
 from pyms.constants import LOGGER_NAME
@@ -18,8 +19,8 @@
 DEFAULT_STATUS_RETRIES = (500, 502, 504)
 
 
-def retry(f):
-    def wrapper(*args, **kwargs):
+def retry(f) -> Any:
+    def wrapper(*args, **kwargs) -> Any:
         response = False
         i = 0
         response_ok = False
@@ -53,7 +54,7 @@ class Service(DriverService):
     }
     tracer = None
 
-    def requests(self, session: requests.Session):
+    def requests(self, session: requests.Session) -> requests.Session:
         """
         A backoff factor to apply between attempts after the second try (most errors are resolved immediately by a
         second try without a delay). urllib3 will sleep for: {backoff factor} * (2 ^ ({number of total retries} - 1))
@@ -113,7 +114,7 @@ def _get_headers(self, headers: dict, propagate_headers: bool = False) -> dict:
         return headers
 
     @staticmethod
-    def _build_url(url, path_params=None):
+    def _build_url(url, path_params: dict = None) -> str:
         """Compose full url replacing placeholders with path_params values.
 
         :param url: base url
@@ -124,7 +125,7 @@ def _build_url(url, path_params=None):
 
         return url.format_map(path_params)
 
-    def parse_response(self, response):
+    def parse_response(self, response: Response) -> dict:
         """Parses response's json object. Checks configuration in order to parse a concrete node or the whole response.
 
         :param response: request's response that contains a valid json
@@ -142,7 +143,8 @@ def parse_response(self, response):
             return {}
 
     @retry
-    def get(self, url, path_params=None, params=None, headers=None, propagate_headers=False, **kwargs):
+    def get(self, url: str, path_params: dict = None, params: dict = None, headers: dict = None,
+            propagate_headers: bool = False, **kwargs) -> Response:
         """Sends a GET request.
 
         :param url: URL for the new :class:`Request` object. Could contain path parameters
@@ -167,7 +169,8 @@ def get(self, url, path_params=None, params=None, headers=None, propagate_header
 
         return response
 
-    def get_for_object(self, url, path_params=None, params=None, headers=None, **kwargs):
+    def get_for_object(self, url: str, path_params: dict = None, params: dict = None, headers: dict = None,
+                       **kwargs) -> dict:
         """Sends a GET request and returns the json representation found in response's content data node.
 
         :param url: URL for the new :class:`Request` object. Could contain path parameters
@@ -184,7 +187,8 @@ def get_for_object(self, url, path_params=None, params=None, headers=None, **kwa
         return self.parse_response(response)
 
     @retry
-    def post(self, url, path_params=None, data=None, json=None, headers=None, **kwargs):
+    def post(self, url: str, path_params: dict = None, data: dict = None, json: dict = None, headers: dict = None,
+             **kwargs) -> Response:
         """Sends a POST request.
 
         :param url: URL for the new :class:`Request` object. Could contain path parameters
@@ -210,7 +214,8 @@ def post(self, url, path_params=None, data=None, json=None, headers=None, **kwar
 
         return response
 
-    def post_for_object(self, url, path_params=None, data=None, json=None, headers=None, **kwargs):
+    def post_for_object(self, url: str, path_params: dict = None, data: dict = None, json: dict = None,
+                        headers: dict = None, **kwargs) -> dict:
         """Sends a POST request and returns the json representation found in response's content data node.
 
         :param url: URL for the new :class:`Request` object. Could contain path parameters
@@ -228,7 +233,7 @@ def post_for_object(self, url, path_params=None, data=None, json=None, headers=N
         return self.parse_response(response)
 
     @retry
-    def put(self, url, path_params=None, data=None, headers=None, **kwargs):
+    def put(self, url: str, path_params: dict = None, data: dict = None, headers: dict = None, **kwargs) -> Response:
         """Sends a PUT request.
 
         :param url: URL for the new :class:`Request` object. Could contain path parameters
@@ -254,7 +259,8 @@ def put(self, url, path_params=None, data=None, headers=None, **kwargs):
 
         return response
 
-    def put_for_object(self, url, path_params=None, data=None, headers=None, **kwargs):
+    def put_for_object(self, url: str, path_params: dict = None, data: dict = None, headers: dict = None,
+                       **kwargs) -> dict:
         """Sends a PUT request and returns the json representation found in response's content data node.
 
         :param url: URL for the new :class:`Request` object. Could contain path parameters
@@ -272,7 +278,7 @@ def put_for_object(self, url, path_params=None, data=None, headers=None, **kwarg
         return self.parse_response(response)
 
     @retry
-    def patch(self, url, path_params=None, data=None, headers=None, **kwargs):
+    def patch(self, url: str, path_params: dict = None, data: dict = None, headers: dict = None, **kwargs) -> Response:
         """Sends a PATCH request.
 
         :param url: URL for the new :class:`Request` object. Could contain path parameters
@@ -298,7 +304,8 @@ def patch(self, url, path_params=None, data=None, headers=None, **kwargs):
 
         return response
 
-    def patch_for_object(self, url, path_params=None, data=None, headers=None, **kwargs):
+    def patch_for_object(self, url: str, path_params: dict = None, data: dict = None, headers: dict = None,
+                         **kwargs) -> dict:
         """Sends a PATCH request and returns the json representation found in response's content data node.
 
         :param url: URL for the new :class:`Request` object. Could contain path parameters
@@ -316,7 +323,7 @@ def patch_for_object(self, url, path_params=None, data=None, headers=None, **kwa
         return self.parse_response(response)
 
     @retry
-    def delete(self, url, path_params=None, headers=None, **kwargs):
+    def delete(self, url: str, path_params: dict = None, headers: dict = None, **kwargs) -> Response:
         """Sends a DELETE request.
 
         :param url: URL for the new :class:`Request` object. Could contain path parameters

pyms/flask/services/swagger.py

@@ -2,6 +2,8 @@
 from pathlib import Path
 from typing import Dict, Any
 
+from flask import Flask
+
 import connexion
 from connexion.resolver import RestyResolver
 
@@ -33,7 +35,7 @@ def get_bundled_specs(main_file: Path) -> Dict[str, Any]:
     return parser.specification
 
 
-def merge_swagger_file(main_file: str):
+def merge_swagger_file(main_file: str) -> None:
     """
     Generate swagger into a single file
     :param main_file: Swagger file path
@@ -71,14 +73,14 @@ class Service(DriverService):
     }
 
     @staticmethod
-    def _get_application_root(config):
+    def _get_application_root(config) -> str:
         try:
             application_root = config.APPLICATION_ROOT
         except AttrDoesNotExistException:
             application_root = "/"
         return application_root
 
-    def init_app(self, config, path):
+    def init_app(self, config, path: Path) -> Flask:
         """
         Initialize Connexion App. See more info in [Connexion Github](https://github.com/zalando/connexion)
         :param config: The Flask configuration defined in the config.yaml:

pyms/flask/services/tracer.py

@@ -1,4 +1,6 @@
 import logging
+from typing import Union
+
 
 try:
     import opentracing
@@ -28,7 +30,7 @@
 DEFAULT_CLIENT = JAEGER_CLIENT
 
 
-def inject_span_in_headers(headers):
+def inject_span_in_headers(headers: dict) -> dict:
     if has_request_context():
         # FLASK https://github.com/opentracing-contrib/python-flask
         tracer = current_app.tracer if getattr(current_app, "tracer") else None
@@ -56,7 +58,7 @@ class Service(DriverService):
         "client": DEFAULT_CLIENT,
     }
 
-    def get_client(self):
+    def get_client(self) -> Union[bool, type]:
         opentracing_tracer = False
         if self.config.client == JAEGER_CLIENT:
             opentracing_tracer = self.init_jaeger_tracer()