refactor: validation at the DAOs level

superset/commands/dataset/exceptions.py

@@ -90,18 +90,6 @@ def __init__(self) -> None:
         super().__init__([_("One or more columns already exist")], field_name="columns")
 
 
-class DatasetColumnsConstraintsValidationError(ValidationError):
-    """
-    Marshmallow validation error when dataset columns have an invalid python_date_format
-    """
-
-    def __init__(self) -> None:
-        super().__init__(
-            [_("One or more columns have an invalid python_date_format")],
-            field_name="columns",
-        )
-
-
 class DatasetMetricsNotFoundValidationError(ValidationError):
     """
     Marshmallow validation error when dataset metric for update does not exist

superset/commands/dataset/update.py

@@ -26,7 +26,6 @@
 from superset.commands.dataset.exceptions import (
     DatabaseChangeValidationError,
     DatasetColumnNotFoundValidationError,
-    DatasetColumnsConstraintsValidationError,
     DatasetColumnsDuplicateValidationError,
     DatasetColumnsExistsValidationError,
     DatasetExistsValidationError,
@@ -140,13 +139,6 @@ def _validate_columns(
                 ):
                     exceptions.append(DatasetColumnsExistsValidationError())
 
-            # validate python_date_format is ISO8601 format
-            for col in columns:
-                if not DatasetDAO.validate_column_pdf_is_iso8601(
-                    col.get("python_date_format", None)
-                ):
-                    exceptions.append(DatasetColumnsConstraintsValidationError())
-
     def _validate_metrics(
         self, metrics: list[dict[str, Any]], exceptions: list[ValidationError]
     ) -> None:

superset/daos/dataset.py

@@ -25,6 +25,7 @@
 
 from superset.connectors.sqla.models import SqlaTable, SqlMetric, TableColumn
 from superset.daos.base import BaseDAO
+from superset.daos.exceptions import DAOUpdateFailedError
 from superset.extensions import db
 from superset.models.core import Database
 from superset.models.dashboard import Dashboard
@@ -153,8 +154,8 @@ def validate_metrics_uniqueness(dataset_id: int, metrics_names: list[str]) -> bo
         return len(dataset_query) == 0
 
     @staticmethod
-    def validate_column_pdf_is_iso8601(dt_format: str) -> bool:
-        if dt_format in ("epoch_s", "epoch_ms", None):
+    def validate_column_python_date_format(dt_format: str) -> bool:
+        if dt_format in ("epoch_s", "epoch_ms"):
             return True
         try:
             dt_str = datetime.now().strftime(dt_format)
@@ -206,6 +207,18 @@ def update_columns(
         then we delete.
         """
 
+        for column in property_columns:
+            if (
+                "python_date_format" in column
+                and column["python_date_format"] is not None
+            ):
+                if not DatasetDAO.validate_column_python_date_format(
+                    column["python_date_format"]
+                ):
+                    raise DAOUpdateFailedError(
+                        "python_date_format is an invalid date/timestamp format."
+                    )
+
         if override_columns:
             db.session.query(TableColumn).filter(
                 TableColumn.table_id == model.id

tests/unit_tests/config_test.py

@@ -211,6 +211,7 @@ def test_python_date_format_by_column_name(
         "dttm_columns": {
             "id": {"python_date_format": "epoch_ms"},
             "dttm": {"python_date_format": "epoch_s"},
+            "duration_ms": {"python_date_format": "invalid"},
         },
     }
     mocker.patch(
@@ -227,6 +228,7 @@ def test_python_date_format_by_column_name(
         return_value=[
             {"column_name": "id", "type": "INTEGER", "is_dttm": False},
             {"column_name": "dttm", "type": "INTEGER", "is_dttm": False},
+            {"column_name": "duration_ms", "type": "INTEGER", "is_dttm": False},
         ],
     )
 
@@ -240,6 +242,12 @@ def test_python_date_format_by_column_name(
     assert dttm_col.is_dttm
     assert dttm_col.python_date_format == "epoch_s"
 
+    duration_ms_col = [c for c in test_table.columns if c.column_name == "duration_ms"][
+        0
+    ]
+    assert duration_ms_col.is_dttm
+    assert duration_ms_col.python_date_format == "invalid"
+
 
 def test_expression_by_column_name(
     mocker: MockerFixture,