feat: add metric storage framework

.github/workflows/ci.yml

@@ -20,15 +20,69 @@ jobs:
         with:
           python-version: 3.9
       - uses: pre-commit/[email protected]
-
-  pytest:
-    name: Backend Pytest Job
+  pytest_unit:
+    name: Pytest Unit Tests with all supported python versions
     strategy:
       matrix:
         python-version: [ "3.8" ,"3.9", "3.10" ]
     runs-on: ubuntu-latest
     needs:
       - style
+    steps:
+      #----------------------------------------------
+      #       check-out repo and set-up python
+      #----------------------------------------------
+      - uses: actions/checkout@v2
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v2
+        with:
+          python-version: ${{ matrix.python-version }}
+      #----------------------------------------------
+      #  -----  install & configure poetry  -----
+      #----------------------------------------------
+      - name: Install Poetry
+        uses: snok/install-poetry@v1
+        with:
+          version: 1.4.1
+          virtualenvs-create: true
+          virtualenvs-in-project: true
+          installer-parallel: true
+      #----------------------------------------------
+      #       load cached venv if cache exists
+      #----------------------------------------------
+      - name: Load cached venv
+        id: cached-poetry-dependencies
+        uses: actions/cache@v2
+        with:
+          path: .venv
+          key: venv-${{ runner.os }}-${{ steps.setup-python.outputs.python-version }}-${{ hashFiles('**/poetry.lock') }}
+      #----------------------------------------------
+      # install dependencies if cache does not exist
+      #----------------------------------------------
+      - name: Install dependencies
+        if: steps.cached-poetry-dependencies.outputs.cache-hit != 'true'
+        run: poetry install --no-interaction --no-root
+      #----------------------------------------------
+      # install your root project, if required
+      #----------------------------------------------
+      - name: Install library
+        run: poetry install --no-interaction --all-extras
+      #-----------------------------------------
+      # Run Pytest
+      #---------------------------------------
+      - name: Run tests
+        run: |
+          source .venv/bin/activate
+          pytest -p no:warnings --ignore=tests/integration/
+
+  pytest:
+    name: Pytest with Codecov
+    strategy:
+      matrix:
+        python-version: [ "3.8" ]
+    runs-on: ubuntu-latest
+    needs:
+      - pytest_unit
     steps:
       #----------------------------------------------
       #       check-out repo and set-up python

Makefile

@@ -4,6 +4,17 @@ pytest:
 	set -x
 	pytest -p no:warnings ./tests/*
 
+pytest_unit:
+	set -e
+	set -x
+	pytest -p no:warnings --ignore=tests/integration/
+
+pytest_integration:
+	set -e
+	set -x
+	pytest -p no:warnings ./tests/integration/*
+
+
 pytest_with_coverage:
 	set -e
 	set -x

datachecks/cli/cli.py

@@ -51,11 +51,12 @@ def main():
     default=None,
     help="Specify the file path for configuration",
 )
-@click.option(
-    "--auto-profile",
-    is_flag=True,
-    help="Specify if the inspection should do auto-profile of all data sources",
-)
+# Disabled for now
+# @click.option(
+#     "--auto-profile",
+#     is_flag=True,
+#     help="Specify if the inspection should do auto-profile of all data sources",
+# )
 @click.option(
     "--html-report",
     is_flag=True,
@@ -69,7 +70,7 @@ def main():
 )
 def inspect(
     config_path: Union[str, None],
-    auto_profile: bool = False,
+    # auto_profile: bool = False, # Disabled for now
     html_report: bool = False,
     report_path: str = "datachecks_report.html",
 ):
@@ -84,7 +85,8 @@ def inspect(
             )
         configuration: Configuration = load_configuration(config_path)
 
-        inspector = Inspect(configuration=configuration, auto_profile=auto_profile)
+        # inspector = Inspect(configuration=configuration, auto_profile=auto_profile) # Disabled for now
+        inspector = Inspect(configuration=configuration)
 
         print("Starting [bold blue]datachecks[/bold blue] inspection...", ":zap:")
         output: InspectOutput = inspector.run()
@@ -144,7 +146,6 @@ def _build_metric_cli_table(*, inspect_output: InspectOutput):
 
 
 def _build_html_report(*, inspect_output: InspectOutput, report_path: str):
-    logger.info(inspect_output)
     template_params = TemplateParams(
         dashboard_id="dcs_dashboard_" + str(uuid.uuid4()).replace("-", ""),
         dashboard_info=DashboardInfoBuilder(inspect_output).build(),

datachecks/core/common/errors.py

@@ -12,11 +12,20 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
+ERROR_RUNTIME = "runtime_error"
 ERROR_CONFIGURATION = "configuration_error"
 ERROR_DATA_SOURCES_CONNECTION = "data_sources_connection_error"
 ERROR_METRIC_GENERATION = "metric_generation_error"
 
 
+class DataChecksRuntimeError(Exception):
+    """Raised when there is an error in the configuration file."""
+
+    def __init__(self, message):
+        super().__init__(message)
+        self.error_code = ERROR_RUNTIME
+
+
 class DataChecksConfigurationError(Exception):
     """Raised when there is an error in the configuration file."""
 

datachecks/core/common/models/configuration.py

@@ -96,6 +96,33 @@ def __post_init__(self):
             )
 
 
+class MetricStorageType(str, Enum):
+    """
+    Metric storage type
+    """
+
+    LOCAL_FILE = "local_file"
+
+
+@dataclass
+class LocalFileStorageParameters:
+    """
+    Local file metric storage parameters
+    """
+
+    path: str
+
+
+@dataclass
+class MetricStorageConfiguration:
+    """
+    Metric storage configuration
+    """
+
+    type: MetricStorageType
+    params: Union[LocalFileStorageParameters]
+
+
 @dataclass
 class Configuration:
     """
@@ -104,3 +131,4 @@ class Configuration:
 
     data_sources: Dict[str, DataSourceConfiguration]
     metrics: Dict[str, MetricConfiguration]
+    storage: Optional[MetricStorageConfiguration] = None

datachecks/core/common/models/metric.py

@@ -11,11 +11,17 @@
 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
-
-from dataclasses import dataclass
+import json
+from dataclasses import asdict, dataclass
+from datetime import datetime
 from enum import Enum
 from typing import Any, Dict, List, Optional, Union
 
+import pytz
+from dateutil import parser
+
+from datachecks.core.utils.utils import EnhancedJSONEncoder
+
 
 class MetricsType(str, Enum):
     """
@@ -56,7 +62,7 @@ class MetricValue:
     identity: str
     value: Union[int, float]
     metric_type: MetricsType
-    timestamp: str
+    timestamp: datetime
     data_source: Optional[str] = None
     expression: Optional[str] = None
     table_name: Optional[str] = None
@@ -66,6 +72,29 @@ class MetricValue:
     reason: Optional[str] = None
     tags: Dict[str, str] = None
 
+    @property
+    def json(self):
+        return json.dumps(asdict(self), cls=EnhancedJSONEncoder)
+
+    @classmethod
+    def from_json(cls, json_string: str):
+        json_obj = json.loads(json_string)
+        parsed_date = parser.parse(json_obj.get("timestamp")).astimezone(tz=pytz.UTC)
+        return cls(
+            identity=json_obj.get("identity"),
+            value=json_obj.get("value"),
+            metric_type=MetricsType(json_obj.get("metric_type")),
+            timestamp=parsed_date,
+            data_source=json_obj.get("data_source", None),
+            expression=json_obj.get("expression", None),
+            table_name=json_obj.get("table_name", None),
+            index_name=json_obj.get("index_name", None),
+            field_name=json_obj.get("field_name", None),
+            is_valid=json_obj.get("is_valid", None),
+            reason=json_obj.get("reason", None),
+            tags=json_obj.get("tags", None),
+        )
+
 
 @dataclass
 class TableMetrics:
@@ -80,6 +109,11 @@ class TableMetrics:
     """
     metrics: Dict[str, MetricValue]
 
+    """
+    Historical values of the metrics is a dictionary of metric identifier and list of metric values
+    """
+    historical_metrics: Optional[Dict[str, List[MetricValue]]] = None
+
 
 @dataclass
 class IndexMetrics:
@@ -94,6 +128,11 @@ class IndexMetrics:
     """
     metrics: Dict[str, MetricValue]
 
+    """
+    Historical values of the metrics is a dictionary of metric identifier and list of metric values
+    """
+    historical_metrics: Optional[Dict[str, List[MetricValue]]] = None
+
 
 @dataclass
 class DataSourceMetrics:

datachecks/core/configuration/configuration_parser.py

@@ -24,8 +24,11 @@
     DataSourceConfiguration,
     DataSourceConnectionConfiguration,
     DataSourceType,
+    LocalFileStorageParameters,
     MetricConfiguration,
     MetricsFilterConfiguration,
+    MetricStorageConfiguration,
+    MetricStorageType,
 )
 from datachecks.core.common.models.data_source_resource import Field, Index, Table
 from datachecks.core.common.models.metric import MetricsType
@@ -46,7 +49,7 @@
 
 
 def parse_data_source_yaml_configurations(
-    data_source_yaml_configurations: List[dict],
+    data_source_yaml_configurations: List[Dict],
 ) -> Dict[str, DataSourceConfiguration]:
     data_source_configurations: Dict[str, DataSourceConfiguration] = {}
     for data_source_yaml_configuration in data_source_yaml_configurations:
@@ -114,7 +117,7 @@ def _parse_threshold_str(threshold: str) -> Threshold:
         )
 
 
-def _parse_validation_configuration(validation_config: dict) -> Validation:
+def _parse_validation_configuration(validation_config: Dict) -> Validation:
     if "threshold" in validation_config:
         threshold = _parse_threshold_str(threshold=validation_config["threshold"])
         return Validation(threshold=threshold)
@@ -153,9 +156,31 @@ def _metric_resource_parser(
             return _parse_resource_field(resource_str, "table")
 
 
+def parse_storage_configurations(
+    storage_yaml_configurations: Dict,
+) -> Union[MetricStorageConfiguration, None]:
+    if storage_yaml_configurations["type"] == "local_file":
+        if "params" not in storage_yaml_configurations:
+            raise DataChecksConfigurationError(
+                "storage params should be provided for local file storage configuration"
+            )
+        if "path" not in storage_yaml_configurations["params"]:
+            raise DataChecksConfigurationError(
+                "path should be provided for local file storage configuration"
+            )
+        return MetricStorageConfiguration(
+            type=MetricStorageType.LOCAL_FILE,
+            params=LocalFileStorageParameters(
+                path=storage_yaml_configurations["params"]["path"]
+            ),
+        )
+    else:
+        return None
+
+
 def parse_metric_configurations(
     data_source_configurations: Dict[str, DataSourceConfiguration],
-    metric_yaml_configurations: List[dict],
+    metric_yaml_configurations: List[Dict],
 ) -> Dict[str, MetricConfiguration]:
     metric_configurations: Dict[str, MetricConfiguration] = {}
 
@@ -231,9 +256,12 @@ def load_configuration_from_yaml_str(yaml_string: str) -> Configuration:
             data_source_configurations=data_source_configurations,
             metric_yaml_configurations=config_dict["metrics"],
         )
-        return Configuration(
+        configuration = Configuration(
             data_sources=data_source_configurations, metrics=metric_configurations
         )
+        if "storage" in config_dict:
+            configuration.storage = parse_storage_configurations(config_dict["storage"])
+        return configuration
     except Exception as ex:
         raise DataChecksConfigurationError(
             message=f"Failed to parse configuration: {str(ex)}"

datachecks/core/datasource/search_datasource.py

@@ -12,7 +12,7 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Dict, List
 
 from dateutil import parser
@@ -219,7 +219,7 @@ def query_get_time_diff(self, index_name: str, field: str) -> int:
             last_updated = response["hits"]["hits"][0]["_source"][field]
 
             last_updated = parser.parse(timestr=last_updated).timestamp()
-            now = datetime.utcnow().timestamp()
+            now = datetime.now(timezone.utc).timestamp()
             return int(now - last_updated)
 
         return 0