source: db: Source using the database abstraction

Fixes: #402

Author: theoilie

CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Docstrings and doctestable examples to `record.py`.
 - Inputs can be validated using operations
   - `validate` parameter in `Input` takes `Operation.instance_name`
+- New db source can utilize any database that inherits from `BaseDatabase`
 - Logistic Regression with SAG optimizer
 - Test tensorflow DNNEstimator documentation exaples in CI
 - Add python code for tensorflow DNNEstimator

dffml/base.py

@@ -336,7 +336,7 @@ class BaseConfigurable(metaclass=BaseConfigurableMetaClass):
     only parameter to the __init__ of a BaseDataFlowFacilitatorObject.
     """
 
-    def __init__(self, config: BaseConfig) -> None:
+    def __init__(self, config: Type[BaseConfig]) -> None:
         """
         BaseConfigurable takes only one argument to __init__,
         its config, which should inherit from BaseConfig. It shall be a object
@@ -538,7 +538,7 @@ class BaseDataFlowFacilitatorObject(
     >>> asyncio.run(main())
     """
 
-    def __init__(self, config: BaseConfig) -> None:
+    def __init__(self, config: Type[BaseConfig]) -> None:
         BaseConfigurable.__init__(self, config)
         # TODO figure out how to call these in __new__
         self.__ensure_property("CONTEXT")

dffml/db/base.py

@@ -9,7 +9,7 @@
 
 
 Condition = collections.namedtuple(
-    "Condtion", ["column", "operation", "value"]
+    "Condition", ["column", "operation", "value"]
 )
 Conditions = Union[List[List[Condition]], List[List[Tuple[str]]]]
 

dffml/source/db.py

@@ -0,0 +1,122 @@
+import collections
+from typing import Type, AsyncIterator, List
+
+from dffml.base import config, BaseConfig
+from dffml.db.base import BaseDatabase, Condition
+from dffml.record import Record
+from dffml.source.source import BaseSource, BaseSourceContext
+from dffml.util.entrypoint import entrypoint
+
+
+@config
+class DbSourceConfig(BaseConfig):
+    db: BaseDatabase
+    table_name: str
+    model_columns: List[str]
+
+
+class DbSourceContext(BaseSourceContext):
+    async def update(self, record: Record):
+        model_columns = self.parent.config.model_columns
+        key_value_pairs = collections.OrderedDict()
+        for key in model_columns:
+            if key.startswith("feature_"):
+                modified_key = key.replace("feature_", "")
+                key_value_pairs[key] = record.data.features[modified_key]
+            elif "_value" in key:
+                target = key.replace("_value", "")
+                if record.data.prediction:
+                    key_value_pairs[key] = record.data.prediction[target][
+                        "value"
+                    ]
+                else:
+                    key_value_pairs[key] = "undetermined"
+            elif "_confidence" in key:
+                target = key.replace("_confidence", "")
+                if record.data.prediction:
+                    key_value_pairs[key] = record.data.prediction[target][
+                        "confidence"
+                    ]
+                else:
+                    key_value_pairs[key] = 1
+            else:
+                key_value_pairs[key] = record.data.__dict__[key]
+        async with self.parent.db() as db_ctx:
+            await db_ctx.insert_or_update(
+                self.parent.config.table_name, key_value_pairs
+            )
+        self.logger.debug("update: %s", await self.record(record.key))
+
+    async def records(self) -> AsyncIterator[Record]:
+        async with self.parent.db() as db_ctx:
+            async for result in db_ctx.lookup(self.parent.config.table_name):
+                yield self.convert_to_record(result)
+
+    def convert_to_record(self, result):
+        modified_record = {
+            "key": "",
+            "data": {"features": {}, "prediction": {}},
+        }
+        for key, value in result.items():
+            if key.startswith("feature_"):
+                modified_record["data"]["features"][
+                    key.replace("feature_", "")
+                ] = value
+            elif ("_value" in key) or ("_confidence" in key):
+                target = key.replace("_value", "").replace("_confidence", "")
+                modified_record["data"]["prediction"][target] = {
+                    "value": result[target + "_value"],
+                    "confidence": result[target + "_confidence"],
+                }
+            else:
+                modified_record[key] = value
+        return Record(modified_record["key"], data=modified_record["data"])
+
+    async def record(self, key: str):
+        record = Record(key)
+        async with self.parent.db() as db_ctx:
+            try:
+                row = await db_ctx.lookup(
+                    self.parent.config.table_name,
+                    cols=None,  # None turns into *. We want all rows
+                    conditions=[[Condition("key", "=", key)]],
+                ).__anext__()
+            except StopAsyncIteration:
+                # This would happen if there is no matching row, so the async generator reached the end
+                return record
+
+        if row is not None:
+            features = {}
+            predictions = {}
+            for key, value in row.items():
+                if key.startswith("feature_"):
+                    features[key.replace("feature_", "")] = value
+                elif "_value" in key:
+                    target = key.replace("_value", "")
+                    predictions[target] = {
+                        "value": row[target + "_value"],
+                        "confidence": row[target + "_confidence"],
+                    }
+            record.merge(
+                Record(
+                    row["key"],
+                    data={"features": features, "prediction": predictions},
+                )
+            )
+        return record
+
+
+@entrypoint("db")
+class DbSource(BaseSource):
+    CONFIG = DbSourceConfig
+    CONTEXT = DbSourceContext
+
+    def __init__(self, cfg: Type[BaseConfig]) -> None:
+        super().__init__(cfg)
+
+    async def __aenter__(self) -> "DbSource":
+        self.db = await self.config.db.__aenter__()
+        return self
+
+    async def __aexit__(self, exc_type, exc_value, traceback):
+        await self.db.__aexit__(exc_type, exc_value, traceback)

dffml/util/entrypoint.py

@@ -81,7 +81,7 @@ def add_entry_point_label(cls):
 def base_entry_point(entrypoint, *args):
     """
     Any class which subclasses from Entrypoint needs this decorator applied to
-    it. The decorator sets the ENTRYPOINT and ENTRY_POINT_NAME proprieties on
+    it. The decorator sets the ENTRYPOINT and ENTRY_POINT_NAME properties on
     the class.
 
     This allows the load() classmethod to be called to load subclasses of the

dffml/util/testing/source.py

@@ -24,7 +24,7 @@ class SourceTest(abc.ABC):
     """
 
     @abc.abstractmethod
-    async def setUpSource(self, fileobj):
+    async def setUpSource(self):
         pass  # pragma: no cover
 
     async def test_update(self):

docs/contributing/codebase.rst

@@ -6,7 +6,7 @@ Codebase Layout And Notes
 Plugins
 -------
 
-DFFML is plugin based. This means that there the source code for the main
+DFFML is plugin based. This means that the source code for the main
 package ``dffml``, is separate from the source code for many of the things you
 might want to use in conjunction with it. For example, if you wanted to use the
 machine learning models based on scikit, you'd install ``dffml-model-scikit``.

scripts/docs/care

@@ -1,4 +1,4 @@
 dffml.operation dffml dffml_feature_git dffml_operations_binsec dffml_feature_auth
 dffml.model dffml dffml_model_tensorflow dffml_model_tensorflow_hub dffml_model_transformers dffml_model_scratch dffml_model_scikit
 dffml.source dffml dffml_source_mysql
-dffml.service.cli dffml dffml_service_http
+dffml.service.cli dffml dffml_service_http

source/mysql/tests/test_source.py

@@ -48,7 +48,7 @@ def setUpClass(cls):
             ca=cls.ca,
         )
         # Make it so that when the client tries to connect to mysql.unittest the
-        # address it get's back is the one for the container
+        # address it gets back is the one for the container
         cls.exit_stack.enter_context(
             patch(
                 "socket.getaddrinfo",

tests/source/test_db.py

@@ -0,0 +1,80 @@
+import os
+import sqlite3
+import tempfile
+from typing import Dict
+
+from dffml.db.sqlite import SqliteDatabaseConfig, SqliteDatabase
+from dffml.util.asynctestcase import AsyncTestCase
+from dffml.util.testing.source import SourceTest
+from dffml.source.db import DbSource, DbSourceConfig
+
+
+class TestDbSource(AsyncTestCase, SourceTest):
+    db_config: SqliteDatabaseConfig
+    source_config: DbSourceConfig
+    database_name: str
+    cols: Dict[str, str]
+
+    SQL_TEARDOWN = """
+    DROP TABLE IF EXISTS `TestTable`;
+    """
+    SQL_SETUP = """
+    CREATE TABLE `TestTable` (
+      `key` varchar(100) NOT NULL,
+      `feature_PetalLength` float DEFAULT NULL,
+      `feature_PetalWidth` float DEFAULT NULL,
+      `feature_SepalLength` float DEFAULT NULL,
+      `feature_SepalWidth` float DEFAULT NULL,
+      `target_name_confidence` float DEFAULT NULL,
+      `target_name_value` varchar(100) DEFAULT NULL,
+      PRIMARY KEY (`key`)
+    );
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        # SQL table info
+        cls.table_name = "TestTable"
+        cls.cols = {
+            "key": "varchar(100) NOT NULL PRIMARY KEY",
+            "feature_PetalLength": "float DEFAULT NULL",
+            "feature_PetalWidth": "float DEFAULT NULL",
+            "feature_SepalLength": "float DEFAULT NULL",
+            "feature_SepalWidth": "float DEFAULT NULL",
+            "target_name_confidence": "float DEFAULT NULL",
+            "target_name_value": "varchar(100) DEFAULT NULL",
+        }
+
+        # Sqlite db file
+        file, cls.database_name = tempfile.mkstemp(suffix=".db")
+        os.close(file)
+
+        # Sqlite config
+        cls.db_config = SqliteDatabaseConfig(cls.database_name)
+
+        # DbSource config
+        cls.source_config = DbSourceConfig(
+            db=SqliteDatabase(cls.db_config),
+            table_name=cls.table_name,
+            model_columns="key feature_PetalLength feature_PetalWidth feature_SepalLength feature_SepalWidth target_name_confidence target_name_value".split(),
+        )
+
+        # Setup connection to reset state (different from the connection used in the tests)
+        conn = sqlite3.connect(cls.database_name)
+        db_cursor = conn.cursor()
+        db_cursor.execute(cls.SQL_TEARDOWN)
+        db_cursor.execute(cls.SQL_SETUP)
+        conn.commit()
+        db_cursor.close()
+        conn.close()
+
+    @classmethod
+    def tearDownClass(cls):
+        os.remove(cls.database_name)
+
+    async def setUpSource(self):
+        return DbSource(self.source_config)
+
+
+# TODO: Potential shortcoming: Is there a way to call this source from the CLI and pass the db object (e.g. SqliteDatabase)?
+# dffml list records -sources primary=dbsource -source-db_implementation sqlite -source-table_name testTable -source-db ??? -source-model_columns "key feature_PetalLength feature_PetalWidth feature_SepalLength feature_SepalWidth target_name_confidence target_name_value"