df: types: Add validate to definition

* Auto validate Inputs via their validate parameter in
   their Definition

Fixes: #349

Author: aghinsa

CHANGELOG.md

@@ -28,6 +28,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - MySQL connector
 - Documented style for imports.
 - Documented use of numpy docstrings.
+- `Inputs` can now be sanitized using function passed in `validate` parameter
 - Helper utilities to take callables with numpy style docstrings and
   create config classes out of them using `make_config`.
 ### Changed

dffml/base.py

@@ -197,7 +197,7 @@ def _fromdict(cls, **kwargs):
                             "config": {
                                 key: value
                                 if is_config_dict(value)
-                                else {"arg": value, "config": {},}
+                                else {"arg": value, "config": {}}
                                 for key, value in config.items()
                             },
                         }

dffml/db/base.py

@@ -2,15 +2,7 @@
 import inspect
 import functools
 import collections
-from typing import (
-    Any,
-    List,
-    Optional,
-    Dict,
-    Tuple,
-    Union,
-    AsyncIterator,
-)
+from typing import Any, List, Optional, Dict, Tuple, Union, AsyncIterator
 
 from dffml.df.base import BaseDataFlowObject, BaseDataFlowObjectContext
 from dffml.util.entrypoint import base_entry_point

dffml/db/sqlite.py

@@ -16,13 +16,13 @@ class SqliteDatabaseConfig:
 
 class SqliteDatabaseContext(SQLDatabaseContext):
     async def create_table(
-        self, table_name: str, cols: Dict[str, str],
+        self, table_name: str, cols: Dict[str, str]
     ) -> None:
         query = self.create_table_query(table_name, cols)
         self.logger.debug(query)
         self.parent.cursor.execute(query)
 
-    async def insert(self, table_name: str, data: Dict[str, Any],) -> None:
+    async def insert(self, table_name: str, data: Dict[str, Any]) -> None:
         query, query_values = self.insert_query(table_name, data)
         async with self.parent.lock:
             with self.parent.db:

dffml/df/exceptions.py

@@ -12,3 +12,7 @@ class DefinitionNotInContext(Exception):
 
 class NotOpImp(Exception):
     pass
+
+
+class InputValidationError(Exception):
+    pass

dffml/df/types.py

@@ -37,6 +37,9 @@ class Definition(NamedTuple):
     lock: bool = False
     # spec is a NamedTuple which could be populated via a dict
     spec: NamedTuple = None
+    # validate property will be a callable (function or lambda) which returns
+    # the sanitized version of the value
+    validate: Callable[[Any], Any] = None
 
     def __repr__(self):
         return self.name
@@ -273,6 +276,8 @@ def __init__(
             parents = []
         if isinstance(value, dict) and definition.spec is not None:
             value = definition.spec(**value)
+        if definition.validate is not None:
+            value = definition.validate(value)
         self.value = value
         self.definition = definition
         self.parents = parents

feature/git/dffml_feature_git/feature/operations.py

@@ -119,7 +119,7 @@ async def git_repo_checkout(repo: Dict[str, str], commit: str):
     await check_output("git", "checkout", commit, cwd=repo.directory)
     return {
         "repo": GitRepoCheckedOutSpec(
-            URL=repo.URL, directory=repo.directory, commit=commit,
+            URL=repo.URL, directory=repo.directory, commit=commit
         )
     }
 

model/scikit/dffml_model_scikit/scikit_base.py

@@ -224,7 +224,7 @@ async def predict(
             predict = np.array(df)
             prediction = predictor(predict)
             self.logger.debug(
-                "Predicted cluster for {}: {}".format(predict, prediction,)
+                "Predicted cluster for {}: {}".format(predict, prediction)
             )
             repo.predicted(prediction[0], self.confidence)
             yield repo

model/scikit/dffml_model_scikit/scikit_models.py

@@ -120,7 +120,7 @@ def applicable_features(self, features):
         QuadraticDiscriminantAnalysis,
         applicable_features,
     ),
-    ("scikitlr", "LinearRegression", LinearRegression, applicable_features,),
+    ("scikitlr", "LinearRegression", LinearRegression, applicable_features),
     (
         "scikitlor",
         "LogisticRegression",
@@ -139,17 +139,12 @@ def applicable_features(self, features):
         ExtraTreesClassifier,
         applicable_features,
     ),
-    (
-        "scikitbgc",
-        "BaggingClassifier",
-        BaggingClassifier,
-        applicable_features,
-    ),
+    ("scikitbgc", "BaggingClassifier", BaggingClassifier, applicable_features),
     ("scikiteln", "ElasticNet", ElasticNet, applicable_features),
-    ("scikitbyr", "BayesianRidge", BayesianRidge, applicable_features,),
-    ("scikitlas", "Lasso", Lasso, applicable_features,),
-    ("scikitard", "ARDRegression", ARDRegression, applicable_features,),
-    ("scikitrsc", "RANSACRegressor", RANSACRegressor, applicable_features,),
+    ("scikitbyr", "BayesianRidge", BayesianRidge, applicable_features),
+    ("scikitlas", "Lasso", Lasso, applicable_features),
+    ("scikitard", "ARDRegression", ARDRegression, applicable_features),
+    ("scikitrsc", "RANSACRegressor", RANSACRegressor, applicable_features),
     ("scikitbnb", "BernoulliNB", BernoulliNB, applicable_features),
     ("scikitmnb", "MultinomialNB", MultinomialNB, applicable_features),
     (
@@ -176,8 +171,8 @@ def applicable_features(self, features):
         OrthogonalMatchingPursuit,
         applicable_features,
     ),
-    ("scikitridge", "Ridge", Ridge, applicable_features,),
-    ("scikitlars", "Lars", Lars, applicable_features,),
+    ("scikitridge", "Ridge", Ridge, applicable_features),
+    ("scikitlars", "Lars", Lars, applicable_features),
     ("scikitkmeans", "KMeans", KMeans, applicable_features),
     ("scikitbirch", "Birch", Birch, applicable_features),
     (

source/mysql/dffml_source_mysql/db.py

@@ -23,7 +23,7 @@ class MySQLDatabaseContext(SQLDatabaseContext):
     BIND_DECLARATION: str = "%s"
 
     async def create_table(
-        self, table_name: str, cols: Dict[str, str],
+        self, table_name: str, cols: Dict[str, str]
     ) -> None:
         query = self.create_table_query(table_name, cols)
         self.logger.debug(query)

source/mysql/dffml_source_mysql/source.py

@@ -51,10 +51,7 @@ async def update(self, repo: Repo):
         self.logger.debug("update: %s", await self.repo(repo.key))
 
     def convert_to_repo(self, result):
-        modified_repo = {
-            "key": "",
-            "data": {"features": {}, "prediction": {}},
-        }
+        modified_repo = {"key": "", "data": {"features": {}, "prediction": {}}}
         for key, value in result.items():
             if key.startswith("feature_"):
                 modified_repo["data"]["features"][

tests/integration/test_service_dev.py

@@ -7,7 +7,6 @@
 import inspect
 import pathlib
 import unittest.mock
-
 from dffml.df.types import DataFlow
 from dffml.cli.cli import CLI
 from dffml.service.dev import Develop

tests/test_types.py

@@ -0,0 +1,82 @@
+from dffml.df.base import op
+from dffml.df.types import DataFlow, Input, Definition
+from dffml.operation.output import GetSingle
+from dffml.util.asynctestcase import AsyncTestCase
+from dffml.df.memory import MemoryOrchestrator
+from dffml.operation.mapping import MAPPING
+from dffml.df.exceptions import InputValidationError
+
+
+def pie_validation(x):
+    if x == 3.14:
+        return x
+    raise InputValidationError()
+
+
+Pie = Definition(name="pie", primitive="float", validate=pie_validation)
+Radius = Definition(name="radius", primitive="float")
+Area = Definition(name="area", primitive="float")
+ShapeName = Definition(
+    name="shape_name", primitive="str", validate=lambda x: x.upper()
+)
+
+
+@op(
+    inputs={"name": ShapeName, "radius": Radius, "pie": Pie},
+    outputs={"shape": MAPPING},
+)
+async def get_circle(name: str, radius: float, pie: float):
+    return {
+        "shape": {
+            "name": name,
+            "radius": radius,
+            "area": pie * radius * radius,
+        }
+    }
+
+
+class TestDefintion(AsyncTestCase):
+    async def setUp(self):
+        self.dataflow = DataFlow(
+            operations={
+                "get_circle": get_circle.op,
+                "get_single": GetSingle.imp.op,
+            },
+            seed=[
+                Input(
+                    value=[get_circle.op.outputs["shape"].name],
+                    definition=GetSingle.op.inputs["spec"],
+                )
+            ],
+            implementations={"get_circle": get_circle.imp},
+        )
+
+    async def test_validate(self):
+        test_inputs = {
+            "area": [
+                Input(value="unitcircle", definition=ShapeName),
+                Input(value=1, definition=Radius),
+                Input(value=3.14, definition=Pie),
+            ]
+        }
+        async with MemoryOrchestrator.withconfig({}) as orchestrator:
+            async with orchestrator(self.dataflow) as octx:
+                async for ctx_str, results in octx.run(test_inputs):
+                    self.assertIn("mapping", results)
+                    results = results["mapping"]
+                    self.assertEqual(results["name"], "UNITCIRCLE")
+                    self.assertEqual(results["area"], 3.14)
+                    self.assertEqual(results["radius"], 1)
+
+    async def test_validation_error(self):
+        with self.assertRaises(InputValidationError):
+            test_inputs = {
+                "area": [
+                    Input(value="unitcircle", definition=ShapeName),
+                    Input(value=1, definition=Radius),
+                    Input(
+                        value=4, definition=Pie
+                    ),  # this should raise validation eror
+                ]
+            }
+            pass

tests/util/double_ret.py

@@ -2,7 +2,7 @@
 
 
 def double_ret(
-    super_cool_arg, *, other_very_cool_arg: Optional[Dict[str, Any]] = None,
+    super_cool_arg, *, other_very_cool_arg: Optional[Dict[str, Any]] = None
 ) -> Tuple[str, Tuple]:
     """
     This is the short description.