Support selecting wildcards in node queries

Author: shangyian

datajunction-server/datajunction_server/internal/namespaces.py

@@ -8,7 +8,7 @@
 
 from sqlalchemy import or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy.orm import joinedload
+from sqlalchemy.orm import joinedload, selectinload
 
 from datajunction_server.api.helpers import get_node_namespace
 from datajunction_server.database.history import ActivityType, EntityType, History
@@ -26,6 +26,7 @@
 )
 from datajunction_server.models.node import NodeMinimumDetail
 from datajunction_server.models.node_type import NodeType
+from datajunction_server.sql.dag import topological_sort
 from datajunction_server.typing import UTCDatetime
 from datajunction_server.utils import SEPARATOR
 
@@ -229,10 +230,10 @@ async def hard_delete_namespace(
     """
     Hard delete a node namespace.
     """
-    node_names = (
+    nodes = (
         (
             await session.execute(
-                select(Node.name)
+                select(Node)
                 .where(
                     or_(
                         Node.namespace.like(
@@ -241,27 +242,33 @@ async def hard_delete_namespace(
                         Node.namespace == namespace,
                     ),
                 )
-                .order_by(Node.name),
+                .order_by(Node.name)
+                .options(
+                    joinedload(Node.current).options(
+                        selectinload(NodeRevision.parents),
+                    ),
+                ),
             )
         )
+        .unique()
         .scalars()
         .all()
     )
 
-    if not cascade and node_names:
+    if not cascade and nodes:
         raise DJActionNotAllowedException(
             message=(
                 f"Cannot hard delete namespace `{namespace}` as there are still the "
-                f"following nodes under it: `{node_names}`. Set `cascade` to true to "
-                "additionally hard delete the above nodes in this namespace. WARNING:"
+                f"following nodes under it: `{[node.name for node in nodes]}`. Set `cascade` to "
+                "true to additionally hard delete the above nodes in this namespace. WARNING:"
                 " this action cannot be undone."
             ),
         )
 
     impacts = {}
-    for node_name in node_names:
-        impacts[node_name] = await hard_delete_node(
-            node_name,
+    for node in reversed(topological_sort(nodes)):
+        impacts[node.name] = await hard_delete_node(
+            node.name,
             session,
             current_user=current_user,
         )

datajunction-server/datajunction_server/internal/nodes.py

@@ -868,7 +868,7 @@ async def propagate_update_downstream(  # pylint: disable=too-many-locals
     )
 
     # The downstreams need to be sorted topologically in order for the updates to be done
-    # in the right order. Otherwise it is possible for a leaf node like a metric to be updated
+    # in the right order. Otherwise, it is possible for a leaf node like a metric to be updated
     # before its upstreams are updated.
     for downstream in downstreams:
         original_node_revision = downstream.current
@@ -1866,18 +1866,17 @@ async def revalidate_node(  # pylint: disable=too-many-locals,too-many-statement
 
     # Check if any columns have been updated
     existing_columns = {col.name: col for col in node.current.columns}  # type: ignore
-    updated_columns = False
+    updated_columns = len(current_node_revision.columns) != len(node_validator.columns)
     for col in node_validator.columns:
         if existing_col := existing_columns.get(col.name):
             if existing_col.type != col.type:
                 existing_col.type = col.type
                 updated_columns = True
         else:
-            node.current.columns.append(col)  # type: ignore  # pragma: no cover
             updated_columns = True  # pragma: no cover
 
     # Only create a new revision if the columns have been updated
-    if updated_columns:  # type: ignore
+    if updated_columns and node_validator.updated_columns:  # type: ignore
         new_revision = copy_existing_node_revision(node.current)  # type: ignore
         new_revision.version = str(
             Version.parse(node.current.version).next_major_version(),  # type: ignore
@@ -1893,16 +1892,16 @@ async def revalidate_node(  # pylint: disable=too-many-locals,too-many-statement
         node_validator.updated_columns = node_validator.modified_columns(
             new_revision,  # type: ignore
         )
-        new_revision.columns = node_validator.columns
 
         # Save the new revision of the child
+        new_revision.columns = node_validator.columns
         node.current_version = new_revision.version  # type: ignore
         new_revision.node_id = node.id  # type: ignore
-        session.add(node)
         session.add(new_revision)
-    await session.commit()
-    await session.refresh(node.current)  # type: ignore
-    await session.refresh(node, ["current"])
+        session.add(node)
+        await session.commit()
+        await session.refresh(node.current)  # type: ignore
+        await session.refresh(node, ["current"])
     return node_validator
 
 
@@ -1918,7 +1917,14 @@ async def hard_delete_node(
     node = await Node.get_by_name(
         session,
         name,
-        options=[joinedload(Node.current), joinedload(Node.revisions)],
+        options=[
+            joinedload(Node.current),
+            joinedload(Node.revisions).options(
+                selectinload(NodeRevision.columns).options(
+                    joinedload(Column.attributes),
+                ),
+            ),
+        ],
         include_inactive=True,
         raise_if_not_exists=False,
     )
@@ -1946,42 +1952,50 @@ async def hard_delete_node(
                 user=current_user.username if current_user else None,
             ),
         )
-        node_validator = await revalidate_node(
-            name=node.name,
-            session=session,
-            current_user=current_user,
-        )
-        impact.append(
-            {
-                "name": node.name,
-                "status": node_validator.status,
-                "effect": "downstream node is now invalid",
-            },
-        )
+        try:
+            node_validator = await revalidate_node(
+                name=node.name,
+                session=session,
+                current_user=current_user,
+            )
+            impact.append(
+                {
+                    "name": node.name,
+                    "status": node_validator.status,
+                    "effect": "downstream node is now invalid",
+                },
+            )
+        except DJNodeNotFound:
+            _logger.warning("Node not found %s", node.name)
 
     # Revalidate all linked nodes
     for node in linked_nodes:
-        session.add(  # Capture this in the downstream node's history
-            History(
-                entity_type=EntityType.LINK,
-                entity_name=name,
-                node=node.name,
-                activity_type=ActivityType.DELETE,
-                user=current_user.username if current_user else None,
-            ),
-        )
-        node_validator = await revalidate_node(
-            name=node.name,
-            session=session,
-            current_user=current_user,
-        )
-        impact.append(
-            {
-                "name": node.name,
-                "status": node_validator.status,
-                "effect": "broken link",
-            },
-        )
+        if node:
+            session.add(  # Capture this in the downstream node's history
+                History(
+                    entity_type=EntityType.LINK,
+                    entity_name=name,
+                    node=node.name,
+                    activity_type=ActivityType.DELETE,
+                    user=current_user.username if current_user else None,
+                ),
+            )
+            try:
+                node_validator = await revalidate_node(
+                    name=node.name,
+                    session=session,
+                    current_user=current_user,
+                    # update=False,
+                )
+                impact.append(
+                    {
+                        "name": node.name,
+                        "status": node_validator.status,
+                        "effect": "broken link",
+                    },
+                )
+            except DJNodeNotFound:
+                _logger.warning("Node not found %s", node.name)
     session.add(  # Capture this in the downstream node's history
         History(
             entity_type=EntityType.NODE,

datajunction-server/datajunction_server/internal/validation.py

@@ -1,12 +1,12 @@
 """Node validation functions."""
 from dataclasses import dataclass, field
-from typing import Dict, List, Set, Union
+from typing import Dict, List, Optional, Set, Union
 
 from sqlalchemy.exc import MissingGreenlet
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from datajunction_server.api.helpers import find_bound_dimensions
-from datajunction_server.database import Column, Node, NodeRevision
+from datajunction_server.database import Column, ColumnAttribute, Node, NodeRevision
 from datajunction_server.errors import DJError, DJException, ErrorCode
 from datajunction_server.models.base import labelize
 from datajunction_server.models.node import NodeRevisionBase, NodeStatus
@@ -22,6 +22,7 @@ class NodeValidator:  # pylint: disable=too-many-instance-attributes
     Node validation
     """
 
+    query_ast: Optional[ast.Query] = None
     status: NodeStatus = NodeStatus.VALID
     columns: List[Column] = field(default_factory=list)
     required_dimensions: List[Column] = field(default_factory=list)
@@ -128,7 +129,12 @@ async def validate_node_data(  # pylint: disable=too-many-locals,too-many-statem
                 name=column_name,
                 display_name=labelize(column_name),
                 type=column_type,
-                attributes=existing_column.attributes if existing_column else [],
+                attributes=[
+                    ColumnAttribute(attribute_type=col_attr.attribute_type)
+                    for col_attr in existing_column.attributes
+                ]
+                if existing_column
+                else [],
                 dimension=existing_column.dimension if existing_column else None,
                 order=idx,
             )