Some adjusts.

Author: moromimay

butterfree/migrations/__init__.py

@@ -1,4 +1,4 @@
 """Holds available migrations."""
-from butterfree.migrations.migrator import Migrator
+from butterfree.migrations.migrate import Migrate
 
-__all__ = ["Migrator"]
+__all__ = ["Migrate"]

butterfree/migrations/database_migration/database_migration.py

@@ -1,6 +1,7 @@
 """Migration entity."""
-
+import logging
 from abc import ABC, abstractmethod
+from itertools import filterfalse
 from typing import Any, Dict, List, Optional
 
 from butterfree.transform import FeatureSet
@@ -45,26 +46,17 @@ def _validate_schema(
         if not db_schema:
             return fs_schema
 
-        for feature in fs_schema:
+        diff_list = list(filterfalse(lambda x: x in db_schema, fs_schema))
+
+        for feature in diff_list:
             matching_features = [
                 x for x in db_schema if x["column_name"] == feature["column_name"]
             ]
 
-            if not matching_features:
-                mismatches.append(feature)
-                continue
-
-            if feature["type"] == matching_features[0]["type"]:
-                continue
+            if matching_features:
+                raise ValueError(f"The {feature['column_name']} can't be changed.")
 
-            mismatches.append(
-                {
-                    "column_name": feature["column_name"],
-                    "type": feature["type"],
-                    "primary_key": feature["primary_key"],
-                    "old_type": matching_features[0]["type"],
-                }
-            )
+            mismatches.append(feature)
 
         return None if mismatches == [] else mismatches
 
@@ -76,12 +68,7 @@ def _get_schema(
         Returns:
             Schema object.
         """
-        try:
-            db_schema = db_client.get_schema(table_name)
-        except RuntimeError:
-            db_schema = None
-
-        return db_schema
+        pass
 
     def run(self, feature_set: FeatureSet) -> None:
         """Runs the migrations.
@@ -90,4 +77,4 @@ def run(self, feature_set: FeatureSet) -> None:
             feature_set: the feature set.
 
         """
-        self._apply_migration(feature_set)
+        pass

butterfree/migrations/migrator.py renamed to butterfree/migrations/migrate.py

@@ -6,7 +6,7 @@
 from butterfree.transform import FeatureSet
 
 
-class Migrator:
+class Migrate:
     """Execute migration operations in a Database based on pipeline Writer.
 
     Attributes:
@@ -37,5 +37,5 @@ def migration(self) -> None:
             self._parse_feature_set_pipeline(pipeline) for pipeline in self.pipelines
         ]
 
-        for migrate, fs in migration_list:
-            migrate.run(fs)
+        for migration, fs in migration_list:
+            migration.run(fs)

tests/unit/butterfree/migrations/database_migration/test_cassandra_migration.py renamed to tests/unit/butterfree/migrations/database_migration/test_database_migration.py

@@ -1,9 +1,10 @@
+import pytest
 from pyspark.sql.types import DoubleType, FloatType, LongType, TimestampType
 
 from butterfree.migrations.database_migration import CassandraMigration
 
 
-class TestCassandraMigration:
+class TestDatabaseMigration:
     def test_validate_schema(self, mocker):
         fs_schema = [
             {"column_name": "id", "type": LongType(), "primary_key": True},
@@ -34,11 +35,6 @@ def test_validate_schema_diff(self, mocker):
         fs_schema = [
             {"column_name": "id", "type": LongType(), "primary_key": True},
             {"column_name": "timestamp", "type": TimestampType(), "primary_key": False},
-            {
-                "column_name": "feature1__avg_over_1_week_rolling_windows",
-                "type": FloatType(),
-                "primary_key": False,
-            },
             {"column_name": "new_feature", "type": FloatType(), "primary_key": False},
         ]
 
@@ -61,15 +57,45 @@ def test_validate_schema_diff(self, mocker):
         m._client = mocker.stub("client")
         schema = m._validate_schema(fs_schema, db_schema)
         assert schema == [
+            {"column_name": "new_feature", "type": FloatType(), "primary_key": False},
+        ]
+
+    def test_validate_schema_diff_invalid(self, mocker):
+        fs_schema = [
+            {"column_name": "id", "type": LongType(), "primary_key": True},
+            {"column_name": "timestamp", "type": TimestampType(), "primary_key": False},
             {
                 "column_name": "feature1__avg_over_1_week_rolling_windows",
                 "type": FloatType(),
                 "primary_key": False,
-                "old_type": DoubleType(),
             },
             {"column_name": "new_feature", "type": FloatType(), "primary_key": False},
         ]
 
+        db_schema = [
+            {"column_name": "id", "type": LongType(), "primary_key": True},
+            {"column_name": "timestamp", "type": TimestampType(), "primary_key": False},
+            {
+                "column_name": "feature1__avg_over_1_week_rolling_windows",
+                "type": DoubleType(),
+                "primary_key": False,
+            },
+            {
+                "column_name": "feature1__avg_over_2_days_rolling_windows",
+                "type": DoubleType(),
+                "primary_key": False,
+            },
+        ]
+
+        with pytest.raises(
+            ValueError,
+            match="The feature1__avg_over_1_week_rolling_windows can't be changed.",
+        ):
+            m = CassandraMigration()
+            m._client = mocker.stub("client")
+
+            m._validate_schema(fs_schema, db_schema)
+
     def test_validate_schema_without_db(self, mocker):
         fs_schema = [
             {"column_name": "id", "type": LongType(), "primary_key": True},