feat: implemented DAG validation with topological sort and tests

.gitignore

@@ -0,0 +1,39 @@
+# Python
+__pycache__/
+*.pyc
+*.pyo
+*.pyd
+*.db
+
+# Virtual environments
+.venv/
+venv/
+env/
+
+# Environment variables
+.env
+
+# Pytest
+.pytest_cache/
+
+# Logs
+*.log
+
+# VS Code
+.vscode/
+
+# OS files
+.DS_Store
+Thumbs.db
+
+# Node (frontend)
+node_modules/
+.next/
+dist/
+build/
+
+# uv
+.uv/
+
+# Misc
+*.sqlite3

backend/app/pipeline/utils.py

@@ -0,0 +1,36 @@
+from collections import defaultdict, deque
+
+def build_graph(pipeline):
+    graph = defaultdict(list)
+    in_degree = defaultdict(int)
+
+    for node in pipeline["nodes"]:
+        node_id = node["id"]
+        graph[node_id] = []
+        in_degree[node_id] = 0
+
+    for edge in pipeline["edges"]:
+        src = edge["source"]
+        dest = edge["target"]
+        if src not in graph or dest not in graph:
+            raise ValueError(f"Invalid edge: {src} -> {dest}")
+        graph[src].append(dest)
+        in_degree[dest] += 1
+
+    return graph, in_degree
+
+def topological_sort(pipeline):
+    graph, in_degree = build_graph(pipeline)
+    queue = deque([node for node in in_degree if in_degree[node] == 0])
+    order = []
+
+    while queue:
+        current = queue.popleft()
+        order.append(current)
+        for neighbor in graph[current]:
+            in_degree[neighbor] -= 1
+            if in_degree[neighbor] == 0:
+                queue.append(neighbor)
+    if len(order) != len(graph):
+        raise ValueError("Cycle detected in the pipeline")
+    return order

backend/pyproject.toml

@@ -9,6 +9,12 @@ dependencies = [
     "numpy>=2.4.4",
     "pandas>=3.0.2",
     "pydantic>=2.12.5",
+    "pytest>=9.0.2",
     "scikit-learn>=1.8.0",
     "uvicorn>=0.43.0",
 ]
+
+[dependency-groups]
+dev = [
+    "pytest>=9.0.2",
+]

backend/tests/test_utils.py

@@ -0,0 +1,51 @@
+from app.pipeline.utils import build_graph, topological_sort
+import pytest
+
+
+def test_build_graph():
+    pipeline = {
+        "nodes": [{"id": "A"}, {"id": "B"}],
+        "edges": [{"source": "A", "target": "B"}]
+    }
+
+    graph, in_degree = build_graph(pipeline)
+
+    assert dict(graph) == {"A": ["B"], "B": []}
+    assert dict(in_degree) == {"A": 0, "B": 1}
+
+
+def test_topological_sort():
+    pipeline = {
+        "nodes": [{"id": "A"}, {"id": "B"}, {"id": "C"}],
+        "edges": [
+            {"source": "A", "target": "B"},
+            {"source": "B", "target": "C"}
+        ]
+    }
+
+    order = topological_sort(pipeline)
+
+    assert order == ["A", "B", "C"]
+
+
+def test_cycle_detection():
+    pipeline = {
+        "nodes": [{"id": "A"}, {"id": "B"}],
+        "edges": [
+            {"source": "A", "target": "B"},
+            {"source": "B", "target": "A"}
+        ]
+    }
+
+    with pytest.raises(ValueError):
+        topological_sort(pipeline)
+
+
+def test_invalid_edge():
+    pipeline = {
+        "nodes": [{"id": "A"}],
+        "edges": [{"source": "A", "target": "X"}]
+    }
+
+    with pytest.raises(ValueError):
+        build_graph(pipeline)