Add typechecker and type annotations

.travis.yml

@@ -14,6 +14,10 @@ jobs:
     stage: lint
     script: make flake8
 
+  - name: MyPy
+    stage: lint
+    script: make mypy
+
   - name: Tests
     stage: test
     script: make test

Makefile

@@ -1,6 +1,6 @@
 SHELL=/bin/bash
 
-.PHONY: build-dev build-prod pylint flake8 lint test
+.PHONY: build-dev build-prod pylint flake8 mypy lint test
 
 build-dev:
 	DEVTOOLS="true" docker-compose build
@@ -14,7 +14,10 @@ pylint: build-dev
 flake8: build-dev
 	docker-compose run --rm --no-deps --entrypoint=python3 api -m flake8 /app/faceanalysis
 
-lint: pylint flake8
+mypy: build-dev
+	docker-compose run --rm --no-deps --entrypoint=python3 api -m mypy /app/faceanalysis
+
+lint: pylint flake8 mypy
 
 test: build-dev
 	$(eval data_dir := $(shell mktemp -d))

app/Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.5-slim
+FROM python:3.6-slim
 
 RUN apt-get -y update \
     && apt-get install -y \

app/faceanalysis/api.py

@@ -1,4 +1,6 @@
 from http import HTTPStatus
+from typing import Tuple
+from typing import Union
 
 from flask import Flask
 from flask import g
@@ -13,6 +15,8 @@
 from faceanalysis import domain
 from faceanalysis.settings import ALLOWED_EXTENSIONS
 
+JsonResponse = Union[dict, Tuple[dict, int]]
+
 app = Flask(__name__)
 app.url_map.strict_slashes = False
 api = Api(app)
@@ -31,13 +35,13 @@
 class AuthenticationToken(Resource):
     method_decorators = [basic_auth.login_required]
 
-    def get(self):
+    def get(self) -> JsonResponse:
         token = auth.generate_auth_token(g.user)
-        return {'token': token.decode('ascii')}
+        return {'token': token}
 
 
 class RegisterUser(Resource):
-    def post(self):
+    def post(self) -> JsonResponse:
         parser = RequestParser()
         parser.add_argument('username',
                             required=True,
@@ -50,7 +54,7 @@ def post(self):
         password = args['password']
 
         try:
-            username = auth.register_user(username, password)
+            auth.register_user(username, password)
         except auth.DuplicateUser:
             return {'error_msg': ERROR_USER_ALREADY_REGISTERED},\
                    HTTPStatus.BAD_REQUEST.value
@@ -61,7 +65,7 @@ def post(self):
 class ProcessImg(Resource):
     method_decorators = [basic_auth.login_required]
 
-    def post(self):
+    def post(self) -> JsonResponse:
         parser = RequestParser()
         parser.add_argument('img_id',
                             required=True,
@@ -80,7 +84,7 @@ def post(self):
 
         return {'img_id': img_id}
 
-    def get(self, img_id):
+    def get(self, img_id: str) -> JsonResponse:
         try:
             status, error = domain.get_processing_status(img_id)
         except domain.ImageDoesNotExist:
@@ -93,7 +97,7 @@ def get(self, img_id):
 class ImgUpload(Resource):
     method_decorators = [basic_auth.login_required]
 
-    def post(self):
+    def post(self) -> JsonResponse:
         parser = RequestParser()
         parser.add_argument('image',
                             type=FileStorage,
@@ -120,22 +124,22 @@ def post(self):
 class ImgMatchList(Resource):
     method_decorators = [basic_auth.login_required]
 
-    def get(self, img_id):
+    def get(self, img_id: str) -> JsonResponse:
         images, distances = domain.lookup_matching_images(img_id)
         return {'imgs': images, 'distances': distances}
 
 
 class ImgList(Resource):
     method_decorators = [basic_auth.login_required]
 
-    def get(self):
+    def get(self) -> JsonResponse:
         images = domain.list_images()
         return {'imgs': images}
 # pylint: enable=no-self-use
 
 
 @basic_auth.verify_password
-def verify_password(username_or_token, password):
+def verify_password(username_or_token: str, password: str) -> bool:
     try:
         user = auth.load_user_from_auth_token(username_or_token)
     except auth.InvalidAuthToken:

app/faceanalysis/auth.py

@@ -29,12 +29,13 @@ class DuplicateUser(AuthError):
     pass
 
 
-def generate_auth_token(user, expiration=TOKEN_EXPIRATION):
-    serializer = Serializer(TOKEN_SECRET_KEY, expires_in=expiration)
-    return serializer.dumps({'id': user.id})
+def generate_auth_token(user: User) -> str:
+    serializer = Serializer(TOKEN_SECRET_KEY, expires_in=TOKEN_EXPIRATION)
+    token = serializer.dumps({'id': user.id})
+    return token.decode('ascii')  # type: ignore
 
 
-def load_user_from_auth_token(token):
+def load_user_from_auth_token(token: str) -> User:
     serializer = Serializer(TOKEN_SECRET_KEY)
     try:
         data = serializer.loads(token)
@@ -50,7 +51,7 @@ def load_user_from_auth_token(token):
     return user
 
 
-def load_user(username, password):
+def load_user(username: str, password: str) -> User:
     db = get_database_manager()
     session = db.get_session()
     user = session.query(User) \
@@ -67,7 +68,7 @@ def load_user(username, password):
     return user
 
 
-def register_user(username, password):
+def register_user(username: str, password: str):
     db = get_database_manager()
     session = db.get_session()
     user = session.query(User) \
@@ -84,4 +85,3 @@ def register_user(username, password):
     session = db.get_session()
     session.add(user)
     db.safe_commit(session)
-    return username

app/faceanalysis/domain.py

@@ -1,3 +1,7 @@
+from typing import IO
+from typing import List
+from typing import Tuple
+
 from faceanalysis import tasks
 from faceanalysis.log import get_logger
 from faceanalysis.models.database_manager import get_database_manager
@@ -26,7 +30,7 @@ class DuplicateImage(FaceAnalysisError):
     pass
 
 
-def process_image(img_id):
+def process_image(img_id: str):
     db = get_database_manager()
     session = db.get_session()
     img_status = session.query(ImageStatus)\
@@ -44,7 +48,7 @@ def process_image(img_id):
     logger.debug('Image %s queued for processing', img_id)
 
 
-def get_processing_status(img_id):
+def get_processing_status(img_id: str) -> Tuple[str, str]:
     db = get_database_manager()
     session = db.get_session()
     img_status = session.query(ImageStatus)\
@@ -59,7 +63,7 @@ def get_processing_status(img_id):
     return img_status.status, img_status.error_msg
 
 
-def upload_image(stream, filename):
+def upload_image(stream: IO[bytes], filename: str) -> str:
     img_id = filename[:filename.find('.')]
     db = get_database_manager()
     session = db.get_session()
@@ -83,7 +87,7 @@ def upload_image(stream, filename):
     return img_id
 
 
-def list_images():
+def list_images() -> List[str]:
     db = get_database_manager()
     session = db.get_session()
     query = session.query(Image)\
@@ -95,7 +99,7 @@ def list_images():
     return image_ids
 
 
-def lookup_matching_images(img_id):
+def lookup_matching_images(img_id: str) -> Tuple[List[str], List[float]]:
     db = get_database_manager()
     session = db.get_session()
     query = session.query(Match)\

app/faceanalysis/face_matcher.py

@@ -1,11 +1,15 @@
 from datetime import datetime
+from typing import List
+from typing import Optional
+from typing import Tuple
 
 import numpy as np
 
-from faceanalysis.face_vectorizer import face_vector_from_text
+from faceanalysis.face_vectorizer import face_vector_from_text, FaceVector
 from faceanalysis.face_vectorizer import face_vector_to_text
 from faceanalysis.face_vectorizer import get_face_vectors
 from faceanalysis.log import get_logger
+from faceanalysis.models.database_manager import Session
 from faceanalysis.models.database_manager import get_database_manager
 from faceanalysis.models.image_status_enum import ImageStatusEnum
 from faceanalysis.models.models import FeatureMapping
@@ -22,22 +26,26 @@
 logger = get_logger(__name__)
 
 
-def _add_entry_to_session(cls, session, **kwargs):
+def _add_entry_to_session(cls, session: Session, **kwargs):
     logger.debug('adding entry to session')
     row = cls(**kwargs)
     session.add(row)
     return row
 
 
-def _store_face_vector(features, img_id, session):
+def _store_face_vector(features: FaceVector, img_id: str, session: Session):
     logger.debug('processing feature mapping')
     _add_entry_to_session(FeatureMapping, session,
                           img_id=img_id,
                           features=face_vector_to_text(features))
     return features
 
 
-def _store_matches(this_img_id, that_img_id, distance_score, session):
+def _store_matches(this_img_id: str,
+                   that_img_id: str,
+                   distance_score: float,
+                   session: Session):
+
     logger.debug('processing matches')
     _add_entry_to_session(Match, session,
                           this_img_id=this_img_id,
@@ -49,7 +57,7 @@ def _store_matches(this_img_id, that_img_id, distance_score, session):
                           distance_score=distance_score)
 
 
-def _load_image_ids_and_face_vectors():
+def _load_image_ids_and_face_vectors() -> Tuple[List[str], np.array]:
     logger.debug('getting all img ids and respective features')
     session = db.get_session()
     known_features = []
@@ -64,7 +72,10 @@ def _load_image_ids_and_face_vectors():
     return img_ids, np.array(known_features)
 
 
-def _prepare_matches(matches, that_img_id, distance_score):
+def _prepare_matches(matches: List[dict],
+                     that_img_id: str,
+                     distance_score: float):
+
     match_exists = False
     for match in matches:
         if match["that_img_id"] == that_img_id:
@@ -79,7 +90,10 @@ def _prepare_matches(matches, that_img_id, distance_score):
         })
 
 
-def _update_img_status(img_id, status=None, error_msg=None):
+def _update_img_status(img_id: str,
+                       status: Optional[ImageStatusEnum] = None,
+                       error_msg: Optional[str] = None):
+
     session = db.get_session()
     update_fields = {}
     if status:
@@ -92,7 +106,9 @@ def _update_img_status(img_id, status=None, error_msg=None):
 
 
 # pylint: disable=len-as-condition
-def _compute_distances(face_encodings, face_to_compare):
+def _compute_distances(face_encodings: np.array,
+                       face_to_compare: FaceVector) -> np.array:
+
     if len(face_encodings) == 0:
         return np.empty(0)
 
@@ -101,7 +117,7 @@ def _compute_distances(face_encodings, face_to_compare):
 # pylint: enable=len-as-condition
 
 
-def process_image(img_id):
+def process_image(img_id: str):
     logger.info('Processing image %s', img_id)
     try:
         img_path = get_image_path(img_id)
@@ -120,7 +136,7 @@ def process_image(img_id):
 
     session = db.get_session()
     _add_entry_to_session(Image, session, img_id=img_id)
-    matches = []
+    matches = []  # type: List[dict]
     for face_vector in face_vectors:
         _store_face_vector(face_vector, img_id, session)
 

app/faceanalysis/face_vectorizer.py

@@ -1,3 +1,4 @@
+from typing import List
 import json
 import os
 
@@ -8,14 +9,15 @@
 from faceanalysis.settings import HOST_DATA_DIR
 from faceanalysis.settings import MOUNTED_DATA_DIR
 
+FaceVector = List[float]
 logger = get_logger(__name__)
 
 
-def _format_mount_path(img_path):
+def _format_mount_path(img_path: str) -> str:
     return '/{}'.format(os.path.basename(img_path))
 
 
-def _format_host_path(img_path):
+def _format_host_path(img_path: str) -> str:
     # volume mounts must be absolute
     if not img_path.startswith('/'):
         img_path = os.path.abspath(img_path)
@@ -28,7 +30,7 @@ def _format_host_path(img_path):
     return img_path
 
 
-def get_face_vectors(img_path, algorithm):
+def get_face_vectors(img_path: str, algorithm: str) -> List[FaceVector]:
     img_mount = _format_mount_path(img_path)
     img_host = _format_host_path(img_path)
     volumes = {img_host: {'bind': img_mount, 'mode': 'ro'}}
@@ -42,9 +44,9 @@ def get_face_vectors(img_path, algorithm):
     return face_vectors[0] if face_vectors else []
 
 
-def face_vector_to_text(vector):
+def face_vector_to_text(vector: FaceVector) -> str:
     return json.dumps(vector)
 
 
-def face_vector_from_text(text):
+def face_vector_from_text(text: str) -> FaceVector:
     return json.loads(text)