subagents

backend/src/models/dataset_review.py

@@ -357,6 +357,8 @@ class SemanticCandidate(Base):
 class FilterSource(str, enum.Enum):
     SUPERSET_NATIVE = "superset_native"
     SUPERSET_URL = "superset_url"
+    SUPERSET_PERMALINK = "superset_permalink"
+    SUPERSET_NATIVE_FILTERS_KEY = "superset_native_filters_key"
     MANUAL = "manual"
     INFERRED = "inferred"
 # [/DEF:FilterSource:Class]

backend/src/models/filter_state.py

@@ -0,0 +1,151 @@
+# [DEF:backend.src.models.filter_state:Module]
+#
+# @COMPLEXITY: 2
+# @SEMANTICS: superset, native, filters, pydantic, models, dataclasses
+# @PURPOSE: Pydantic models for Superset native filter state extraction and restoration.
+# @LAYER: Models
+# @RELATION: [DEPENDS_ON] ->[pydantic]
+
+# [SECTION: IMPORTS]
+from typing import Any, Dict, List, Optional
+from pydantic import BaseModel, ConfigDict, Field
+# [/SECTION]
+
+
+# [DEF:FilterState:Model]
+# @COMPLEXITY: 2
+# @PURPOSE: Represents the state of a single native filter.
+# @DATA_CONTRACT: Input[extraFormData: Dict, filterState: Dict, ownState: Optional[Dict]] -> Model[FilterState]
+class FilterState(BaseModel):
+    """Single native filter state with extraFormData, filterState, and ownState."""
+    
+    model_config = ConfigDict(extra="allow")
+    
+    extraFormData: Dict[str, Any] = Field(default_factory=dict, description="Extra form data for the filter")
+    filterState: Dict[str, Any] = Field(default_factory=dict, description="Current filter state")
+    ownState: Dict[str, Any] = Field(default_factory=dict, description="Own state of the filter")
+# [/DEF:FilterState:Model]
+
+
+# [DEF:NativeFilterDataMask:Model]
+# @COMPLEXITY: 2
+# @PURPOSE: Represents the dataMask containing all native filter states.
+# @DATA_CONTRACT: Input[Dict[filter_id, FilterState]] -> Model[NativeFilterDataMask]
+class NativeFilterDataMask(BaseModel):
+    """Container for all native filter states in a dashboard."""
+    
+    model_config = ConfigDict(extra="allow")
+    
+    filters: Dict[str, Any] = Field(default_factory=dict, description="Map of filter ID to filter state data")
+    
+    def get_filter_ids(self) -> List[str]:
+        """Return list of all filter IDs."""
+        return list(self.filters.keys())
+    
+    def get_extra_form_data(self, filter_id: str) -> Dict[str, Any]:
+        """Get extraFormData for a specific filter."""
+        filter_state = self.filters.get(filter_id)
+        if filter_state:
+            return filter_state.extraFormData
+        return {}
+# [/DEF:NativeFilterDataMask:Model]
+
+
+# [DEF:ParsedNativeFilters:Model]
+# @COMPLEXITY: 2
+# @PURPOSE: Result of parsing native filters from permalink or native_filters_key.
+# @DATA_CONTRACT: Input[dataMask: Dict, metadata: Dict] -> Model[ParsedNativeFilters]
+class ParsedNativeFilters(BaseModel):
+    """Result of extracting native filters from a Superset URL."""
+    
+    model_config = ConfigDict(extra="allow")
+    
+    dataMask: Dict[str, Any] = Field(default_factory=dict, description="Extracted dataMask from filters")
+    filter_type: Optional[str] = Field(default=None, description="Type of filter: permalink, native_filters_key, or native_filters")
+    dashboard_id: Optional[str] = Field(default=None, description="Dashboard ID if available")
+    permalink_key: Optional[str] = Field(default=None, description="Permalink key if used")
+    filter_state_key: Optional[str] = Field(default=None, description="Filter state key if used")
+    active_tabs: List[str] = Field(default_factory=list, description="Active tabs in dashboard")
+    anchor: Optional[str] = Field(default=None, description="Anchor position in dashboard")
+    chart_states: Dict[str, Any] = Field(default_factory=dict, description="Chart states in dashboard")
+    
+    def has_filters(self) -> bool:
+        """Check if any filters were extracted."""
+        return bool(self.dataMask)
+    
+    def get_filter_count(self) -> int:
+        """Get the number of filters extracted."""
+        return len(self.dataMask)
+# [/DEF:ParsedNativeFilters:Model]
+
+
+# [DEF:DashboardURLFilterExtraction:Model]
+# @COMPLEXITY: 2
+# @PURPOSE: Result of parsing a complete dashboard URL for filter information.
+# @DATA_CONTRACT: Input[url: str, dashboard_id: Optional, filter_type: Optional, filters: Dict] -> Model[DashboardURLFilterExtraction]
+class DashboardURLFilterExtraction(BaseModel):
+    """Result of parsing a Superset dashboard URL to extract filter state."""
+    
+    model_config = ConfigDict(extra="allow")
+    
+    url: str = Field(..., description="Original dashboard URL")
+    dashboard_id: Optional[str] = Field(default=None, description="Extracted dashboard ID")
+    filter_type: Optional[str] = Field(default=None, description="Type of filter found")
+    filters: ParsedNativeFilters = Field(default_factory=ParsedNativeFilters, description="Extracted filter data")
+    success: bool = Field(default=True, description="Whether extraction was successful")
+    error: Optional[str] = Field(default=None, description="Error message if extraction failed")
+# [/DEF:DashboardURLFilterExtraction:Model]
+
+
+# [DEF:ExtraFormDataMerge:Model]
+# @COMPLEXITY: 2
+# @PURPOSE: Configuration for merging extraFormData from different sources.
+# @DATA_CONTRACT: Input[append_keys: List[str], override_keys: List[str]] -> Model[ExtraFormDataMerge]
+class ExtraFormDataMerge(BaseModel):
+    """Configuration for merging extraFormData between original and new filter values."""
+    
+    # Keys that should be appended (arrays, filters)
+    append_keys: List[str] = Field(
+        default_factory=lambda: ["filters", "extras", "columns", "metrics"],
+        description="Keys that should be merged by appending"
+    )
+    # Keys that should be overridden (single values)
+    override_keys: List[str] = Field(
+        default_factory=lambda: ["time_range", "time_grain_sqla", "time_column", "granularity"],
+        description="Keys that should be overridden by new values"
+    )
+    
+    def merge(self, original: Dict[str, Any], new: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Merge two extraFormData dictionaries.
+        
+        @param original: Original extraFormData from dashboard metadata
+        @param new: New extraFormData from URL/permalink
+        @return: Merged extraFormData dictionary
+        """
+        result = {}
+        
+        # Start with original
+        for key, value in original.items():
+            result[key] = value
+        
+        # Apply overrides and appends from new
+        for key, new_value in new.items():
+            if key in self.override_keys:
+                # Override the value
+                result[key] = new_value
+            elif key in self.append_keys:
+                # Append to the existing value
+                existing = result.get(key)
+                if isinstance(existing, list) and isinstance(new_value, list):
+                    result[key] = existing + new_value
+                else:
+                    result[key] = new_value
+            else:
+                result[key] = new_value
+        
+        return result
+# [/DEF:ExtraFormDataMerge:Model]
+
+
+# [/DEF:backend.src.models.filter_state:Module]