Enable Mypy static type checking (#749)

* Enable Mypy support

* Remove proto compilation from linting

* Remove broken rebase imports

* Finalize Mypy tests and fix broken imports

* Remove CLI duplication of methods

* Fix failing undefined responses

* Remove comment about requiring protos to be compiled

* Enable proto compilation for Python linting in GH action

Co-authored-by: Tan Yee Jian <tanyeejian@gmail.com>
Co-authored-by: Willem Pienaar <git@willem.co>
Co-authored-by: Julio Anthony Leonard <julioanthonyleonard@gmail.com>

.github/workflows/code_standards.yaml

@@ -18,6 +18,8 @@ jobs:
       - uses: actions/checkout@v2
       - name: install dependencies
         run: make install-python-ci-dependencies
+      - name: compile protos
+        run: make compile-protos-python
       - name: lint python
         run: make lint-python
 

Makefile

@@ -1,12 +1,12 @@
-# 
+#
 #  Copyright 2019 The Feast Authors
-# 
+#
 #  Licensed under the Apache License, Version 2.0 (the "License");
 #  you may not use this file except in compliance with the License.
 #  You may obtain a copy of the License at
-# 
+#
 #      https://www.apache.org/licenses/LICENSE-2.0
-# 
+#
 #  Unless required by applicable law or agreed to in writing, software
 #  distributed under the License is distributed on an "AS IS" BASIS,
 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -73,8 +73,7 @@ format-python:
 	cd ${ROOT_DIR}/sdk/python; black --target-version py37 feast tests
 
 lint-python:
-	# TODO: This mypy test needs to be re-enabled and all failures fixed
-	#cd ${ROOT_DIR}/sdk/python; mypy feast/ tests/
+	cd ${ROOT_DIR}/sdk/python; mypy feast/ tests/
 	cd ${ROOT_DIR}/sdk/python; flake8 feast/ tests/
 	cd ${ROOT_DIR}/sdk/python; black --check feast tests
 
@@ -104,7 +103,7 @@ build-push-docker:
 	@$(MAKE) push-core-docker registry=$(REGISTRY) version=$(VERSION)
 	@$(MAKE) push-serving-docker registry=$(REGISTRY) version=$(VERSION)
 	@$(MAKE) push-ci-docker registry=$(REGISTRY)
-	
+
 build-docker: build-core-docker build-serving-docker build-ci-docker
 
 push-core-docker:

sdk/python/feast/cli.py

@@ -15,6 +15,7 @@
 import json
 import logging
 import sys
+from typing import Dict, List
 
 import click
 import pkg_resources
@@ -120,28 +121,12 @@ def feature():
     pass
 
 
-def _get_features_labels_dict(label_str: str):
-    """
-    Converts CLI input labels string to dictionary format if provided string is valid.
-    """
-    labels_dict = {}
-    labels_kv = label_str.split(",")
-    if label_str == "":
-        return labels_dict
-    if len(labels_kv) % 2 == 1:
-        raise ValueError("Uneven key-value label pairs were entered")
-    for k, v in zip(labels_kv[0::2], labels_kv[1::2]):
-        labels_dict[k] = v
-    return labels_dict
-
-
-def _get_features_entities(entities_str: str):
+def _convert_entity_string_to_list(entities_str: str) -> List[str]:
     """
     Converts CLI input entities string to list format if provided string is valid.
     """
-    entities_list = []
     if entities_str == "":
-        return entities_list
+        return []
     return entities_str.split(",")
 
 
@@ -173,8 +158,8 @@ def feature_list(project: str, entities: str, labels: str):
     """
     feast_client = Client()  # type: Client
 
-    entities_list = _get_features_entities(entities)
-    labels_dict = _get_features_labels_dict(labels)
+    entities_list = _convert_entity_string_to_list(entities)
+    labels_dict: Dict[str, str] = _get_labels_dict(labels)
 
     table = []
     for feature_ref, feature in feast_client.list_features_by_ref(
@@ -195,11 +180,11 @@ def feature_set():
     pass
 
 
-def _get_labels_dict(label_str: str):
+def _get_labels_dict(label_str: str) -> Dict[str, str]:
     """
     Converts CLI input labels string to dictionary format if provided string is valid.
     """
-    labels_dict = {}
+    labels_dict: Dict[str, str] = {}
     labels_kv = label_str.split(",")
     if label_str == "":
         return labels_dict

sdk/python/feast/client.py

@@ -90,7 +90,7 @@
 
 _logger = logging.getLogger(__name__)
 
-CPU_COUNT = os.cpu_count()  # type: int
+CPU_COUNT: int = len(os.sched_getaffinity(0))
 
 
 class Client:
@@ -123,9 +123,9 @@ def __init__(self, options: Optional[Dict[str, str]] = None, **kwargs):
             options = dict()
         self._config = Config(options={**options, **kwargs})
 
-        self._core_service_stub: CoreServiceStub = None
-        self._serving_service_stub: ServingServiceStub = None
-        self._auth_metadata = None
+        self._core_service_stub: Optional[CoreServiceStub] = None
+        self._serving_service_stub: Optional[ServingServiceStub] = None
+        self._auth_metadata: Optional[grpc.AuthMetadataPlugin] = None
 
         # Configure Auth Metadata Plugin if auth is enabled
         if self._config.getboolean(CONFIG_CORE_ENABLE_AUTH_KEY):
@@ -475,7 +475,7 @@ def get_feature_set(
                 raise ValueError("No project has been configured.")
 
         try:
-            get_feature_set_response = self._core_service_stub.GetFeatureSet(
+            get_feature_set_response = self._core_service.GetFeatureSet(
                 GetFeatureSetRequest(project=project, name=name.strip()),
                 metadata=self._get_grpc_metadata(),
             )  # type: GetFeatureSetResponse
@@ -719,9 +719,9 @@ def list_ingest_jobs(
         )
         request = ListIngestionJobsRequest(filter=list_filter)
         # make list request & unpack response
-        response = self._core_service_stub.ListIngestionJobs(request)
+        response = self._core_service_stub.ListIngestionJobs(request)  # type: ignore
         ingest_jobs = [
-            IngestJob(proto, self._core_service_stub) for proto in response.jobs
+            IngestJob(proto, self._core_service_stub) for proto in response.jobs  # type: ignore
         ]
         return ingest_jobs
 
@@ -737,7 +737,7 @@ def restart_ingest_job(self, job: IngestJob):
         """
         request = RestartIngestionJobRequest(id=job.id)
         try:
-            self._core_service_stub.RestartIngestionJob(request)
+            self._core_service.RestartIngestionJob(request)  # type: ignore
         except grpc.RpcError as e:
             raise grpc.RpcError(e.details())
 
@@ -753,7 +753,7 @@ def stop_ingest_job(self, job: IngestJob):
         """
         request = StopIngestionJobRequest(id=job.id)
         try:
-            self._core_service_stub.StopIngestionJob(request)
+            self._core_service.StopIngestionJob(request)  # type: ignore
         except grpc.RpcError as e:
             raise grpc.RpcError(e.details())
 
@@ -817,11 +817,12 @@ def ingest(
         while True:
             if timeout is not None and time.time() - current_time >= timeout:
                 raise TimeoutError("Timed out waiting for feature set to be ready")
-            feature_set = self.get_feature_set(name)
+            fetched_feature_set: Optional[FeatureSet] = self.get_feature_set(name)
             if (
-                feature_set is not None
-                and feature_set.status == FeatureSetStatus.STATUS_READY
+                fetched_feature_set is not None
+                and fetched_feature_set.status == FeatureSetStatus.STATUS_READY
             ):
+                feature_set = fetched_feature_set
                 break
             time.sleep(3)
 
@@ -944,7 +945,7 @@ def get_statistics(
             if end_date is not None:
                 request.end_date.CopyFrom(Timestamp(seconds=int(end_date.timestamp())))
 
-        return self._core_service_stub.GetFeatureStatistics(
+        return self._core_service.GetFeatureStatistics(
             request
         ).dataset_feature_statistics_list
 

sdk/python/feast/feature_set.py

@@ -22,6 +22,7 @@
 from google.protobuf.duration_pb2 import Duration
 from google.protobuf.json_format import MessageToDict, MessageToJson
 from google.protobuf.message import Message
+from google.protobuf.timestamp_pb2 import Timestamp
 from pandas.api.types import is_datetime64_ns_dtype
 from pyarrow.lib import TimestampType
 
@@ -62,20 +63,20 @@ def __init__(
         self._project = project
         self._fields = OrderedDict()  # type: Dict[str, Field]
         if features is not None:
-            self.features = features
+            self.features: Optional[List[Feature]] = features
         if entities is not None:
             self.entities = entities
         if source is None:
             self._source = None
         else:
             self._source = source
         if labels is None:
-            self._labels = OrderedDict()
+            self._labels = OrderedDict()  # type: MutableMapping[str, str]
         else:
             self._labels = labels
         self._max_age = max_age
         self._status = None
-        self._created_timestamp = None
+        self._created_timestamp: Optional[Timestamp] = None
 
     def __eq__(self, other):
         if not isinstance(other, FeatureSet):
@@ -314,7 +315,7 @@ def drop(self, name: str):
         """
         del self._fields[name]
 
-    def _add_fields(self, fields: List[Field]):
+    def _add_fields(self, fields):
         """
         Adds multiple Fields to a Feature Set
 
@@ -379,8 +380,9 @@ def infer_fields_from_df(
 
         # Create dictionary of fields that will not be inferred (manually set)
         provided_fields = OrderedDict()
+        fields = _create_field_list(entities, features)
 
-        for field in entities + features:
+        for field in fields:
             if not isinstance(field, Field):
                 raise Exception(f"Invalid field object type provided {type(field)}")
             if field.name not in provided_fields:
@@ -518,8 +520,9 @@ def infer_fields_from_pa(
 
         # Create dictionary of fields that will not be inferred (manually set)
         provided_fields = OrderedDict()
+        fields = _create_field_list(entities, features)
 
-        for field in entities + features:
+        for field in fields:
             if not isinstance(field, Field):
                 raise Exception(f"Invalid field object type provided {type(field)}")
             if field.name not in provided_fields:
@@ -835,7 +838,7 @@ def from_proto(cls, feature_set_proto: FeatureSetProto):
             if len(feature_set_proto.spec.project) == 0
             else feature_set_proto.spec.project,
         )
-        feature_set._status = feature_set_proto.meta.status
+        feature_set._status = feature_set_proto.meta.status  # type: ignore
         feature_set._created_timestamp = feature_set_proto.meta.created_timestamp
         return feature_set
 
@@ -1041,3 +1044,29 @@ def _infer_pd_column_type(column, series, rows_to_sample):
             dtype = current_dtype
 
     return dtype
+
+
+def _create_field_list(entities: List[Entity], features: List[Feature]) -> List[Field]:
+    """
+    Convert entities and features List to Field List
+
+    Args:
+        entities: List of Entity Objects
+        features: List of Features Objects
+
+
+    Returns:
+         List[Field]:
+            List of field from entities and features combined
+    """
+    fields: List[Field] = []
+
+    for entity in entities:
+        if isinstance(entity, Field):
+            fields.append(entity)
+
+    for feature in features:
+        if isinstance(feature, Field):
+            fields.append(feature)
+
+    return fields