Add dependency validations for DYNAMIC_TABLE, HYBRID_TABLE, TASK, VIEW

snowddl/validator/__init__.py

@@ -1,14 +1,22 @@
 from .abc_validator import AbstractValidator
 from .business_role import BusinessRoleValidator
 from .database import DatabaseValidator
+from .dynamic_table import DynamicTableValidator
+from .hybrid_table import HybridTableValidator
 from .schema import SchemaValidator
+from .task import TaskValidator
 from .technical_role import TechnicalRoleValidator
 from .user import UserValidator
+from .view import ViewValidator
 
 
 default_validate_sequence = [
     DatabaseValidator,
     SchemaValidator,
+    DynamicTableValidator,
+    HybridTableValidator,
+    TaskValidator,
+    ViewValidator,
     TechnicalRoleValidator,
     BusinessRoleValidator,
     UserValidator,

snowddl/validator/dynamic_table.py

@@ -0,0 +1,19 @@
+from snowddl.blueprint import DynamicTableBlueprint
+from snowddl.validator.abc_validator import AbstractValidator
+
+
+class DynamicTableValidator(AbstractValidator):
+    def get_blueprints(self):
+        return self.config.get_blueprints_by_type(DynamicTableBlueprint)
+
+    def validate_blueprint(self, bp: DynamicTableBlueprint):
+        self._validate_depends_on(bp)
+
+    def _validate_depends_on(self, bp: DynamicTableBlueprint):
+        for depends_on_name in bp.depends_on:
+            if depends_on_name == bp.full_name:
+                raise ValueError(f"Dynamic table [{bp.full_name}] cannot depend on itself")
+
+            if depends_on_name not in self.config.get_blueprints_by_type(DynamicTableBlueprint):
+                raise ValueError(f"Dynamic table [{bp.full_name}] depends on another dynamic table "
+                                 f"[{depends_on_name}] which does not exist in config")

snowddl/validator/hybrid_table.py

@@ -0,0 +1,19 @@
+from snowddl.blueprint import HybridTableBlueprint
+from snowddl.validator.abc_validator import AbstractValidator
+
+
+class HybridTableValidator(AbstractValidator):
+    def get_blueprints(self):
+        return self.config.get_blueprints_by_type(HybridTableBlueprint)
+
+    def validate_blueprint(self, bp: HybridTableBlueprint):
+        self._validate_depends_on(bp)
+
+    def _validate_depends_on(self, bp: HybridTableBlueprint):
+        for depends_on_name in bp.depends_on:
+            if depends_on_name == bp.full_name:
+                raise ValueError(f"Hybrid table [{bp.full_name}] cannot depend on itself")
+
+            if depends_on_name not in self.config.get_blueprints_by_type(HybridTableBlueprint):
+                raise ValueError(f"Hybrid table [{bp.full_name}] depends on another hybrid table "
+                                 f"[{depends_on_name}] which does not exist in config")

snowddl/validator/task.py

@@ -0,0 +1,19 @@
+from snowddl.blueprint import TaskBlueprint
+from snowddl.validator.abc_validator import AbstractValidator
+
+
+class TaskValidator(AbstractValidator):
+    def get_blueprints(self):
+        return self.config.get_blueprints_by_type(TaskBlueprint)
+
+    def validate_blueprint(self, bp: TaskBlueprint):
+        self._validate_depends_on(bp)
+
+    def _validate_depends_on(self, bp: TaskBlueprint):
+        for depends_on_name in bp.depends_on:
+            if depends_on_name == bp.full_name:
+                raise ValueError(f"Task [{bp.full_name}] cannot depend on itself")
+
+            if depends_on_name not in self.config.get_blueprints_by_type(TaskBlueprint):
+                raise ValueError(f"Task [{bp.full_name}] depends on another task "
+                                 f"[{depends_on_name}] which does not exist in config")

snowddl/validator/view.py

@@ -0,0 +1,19 @@
+from snowddl.blueprint import ViewBlueprint
+from snowddl.validator.abc_validator import AbstractValidator
+
+
+class ViewValidator(AbstractValidator):
+    def get_blueprints(self):
+        return self.config.get_blueprints_by_type(ViewBlueprint)
+
+    def validate_blueprint(self, bp: ViewBlueprint):
+        self._validate_depends_on(bp)
+
+    def _validate_depends_on(self, bp: ViewBlueprint):
+        for depends_on_name in bp.depends_on:
+            if depends_on_name == bp.full_name:
+                raise ValueError(f"View [{bp.full_name}] cannot depend on itself")
+
+            if depends_on_name not in self.config.get_blueprints_by_type(ViewBlueprint):
+                raise ValueError(f"View [{bp.full_name}] depends on another view "
+                                 f"[{depends_on_name}] which does not exist in config")