Added Hooks implementation
This commit is contained in:
386
src/components/jsonviewer/hooks.py
Normal file
386
src/components/jsonviewer/hooks.py
Normal file
@@ -0,0 +1,386 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from enum import Enum
|
||||
from typing import Any, Callable, Optional
|
||||
|
||||
|
||||
class EventType(Enum):
|
||||
RENDER = "render"
|
||||
CLICK = "click"
|
||||
HOVER = "hover"
|
||||
FOCUS = "focus"
|
||||
|
||||
|
||||
class DefaultEditableTypes:
|
||||
pass
|
||||
|
||||
|
||||
class HookContext:
|
||||
"""Enhanced context object passed to hook executors"""
|
||||
|
||||
def __init__(self, key: Any, node: Any, helper: Any, jsonviewer: Any,
|
||||
json_path: str = None, parent_node: Any = None):
|
||||
self.key = key
|
||||
self.node = node
|
||||
self.helper = helper
|
||||
self.jsonviewer = jsonviewer
|
||||
self.json_path = json_path or ""
|
||||
self.parent_node = parent_node
|
||||
self.metadata = {}
|
||||
|
||||
def get_node_type(self) -> str:
|
||||
"""Returns string representation of node type"""
|
||||
if hasattr(self.node, '__class__'):
|
||||
return self.node.__class__.__name__
|
||||
return type(self.node.value).__name__ if hasattr(self.node, 'value') else "unknown"
|
||||
|
||||
def get_value(self) -> Any:
|
||||
"""Gets the actual value from the node"""
|
||||
return getattr(self.node, 'value', self.node)
|
||||
|
||||
def is_leaf_node(self) -> bool:
|
||||
"""Checks if this is a leaf node (no children)"""
|
||||
return not hasattr(self.node, 'children') or not self.node.children
|
||||
|
||||
|
||||
class Condition(ABC):
|
||||
"""Base class for all conditions"""
|
||||
|
||||
@abstractmethod
|
||||
def evaluate(self, context: HookContext) -> bool:
|
||||
pass
|
||||
|
||||
|
||||
class WhenLongText(Condition):
|
||||
"""Condition: text length > threshold"""
|
||||
|
||||
def __init__(self, threshold: int = 100):
|
||||
self.threshold = threshold
|
||||
|
||||
def evaluate(self, context: HookContext) -> bool:
|
||||
value = context.get_value()
|
||||
return isinstance(value, str) and len(value) > self.threshold
|
||||
|
||||
|
||||
class WhenEditable(Condition):
|
||||
"""Condition: node is editable (configurable logic)"""
|
||||
|
||||
def __init__(self, editable_paths: list[str] = None, editable_types: list[type] = DefaultEditableTypes):
|
||||
self.editable_paths = set(editable_paths or [])
|
||||
if editable_types is None:
|
||||
self.editable_types = set()
|
||||
else:
|
||||
self.editable_types = set([str, int, float, bool] if editable_types is DefaultEditableTypes else editable_types)
|
||||
|
||||
def evaluate(self, context: HookContext) -> bool:
|
||||
# Check if path is in editable paths
|
||||
if self.editable_paths and context.json_path in self.editable_paths:
|
||||
return True
|
||||
|
||||
# Check if type is editable
|
||||
value = context.get_value()
|
||||
return type(value) in self.editable_types and context.is_leaf_node()
|
||||
|
||||
|
||||
class WhenType(Condition):
|
||||
"""Condition: node value is of specific type"""
|
||||
|
||||
def __init__(self, target_type: type):
|
||||
self.target_type = target_type
|
||||
|
||||
def evaluate(self, context: HookContext) -> bool:
|
||||
value = context.get_value()
|
||||
return isinstance(value, self.target_type)
|
||||
|
||||
|
||||
class WhenKey(Condition):
|
||||
"""Condition: node key matches pattern"""
|
||||
|
||||
def __init__(self, key_pattern: Any):
|
||||
self.key_pattern = key_pattern
|
||||
|
||||
def evaluate(self, context: HookContext) -> bool:
|
||||
if callable(self.key_pattern):
|
||||
return self.key_pattern(context.key)
|
||||
return context.key == self.key_pattern
|
||||
|
||||
|
||||
class WhenPath(Condition):
|
||||
"""Condition: JSON path matches pattern"""
|
||||
|
||||
def __init__(self, path_pattern: str):
|
||||
self.path_pattern = path_pattern
|
||||
|
||||
def evaluate(self, context: HookContext) -> bool:
|
||||
import re
|
||||
return bool(re.match(self.path_pattern, context.json_path))
|
||||
|
||||
|
||||
class WhenValue(Condition):
|
||||
"""Condition: node value matches specific value or predicate"""
|
||||
|
||||
def __init__(self, target_value: Any = None, predicate: Callable[[Any], bool] = None):
|
||||
if target_value is not None and predicate is not None:
|
||||
raise ValueError("Cannot specify both target_value and predicate")
|
||||
if target_value is None and predicate is None:
|
||||
raise ValueError("Must specify either target_value or predicate")
|
||||
|
||||
self.target_value = target_value
|
||||
self.predicate = predicate
|
||||
|
||||
def evaluate(self, context: HookContext) -> bool:
|
||||
value = context.get_value()
|
||||
|
||||
if self.predicate:
|
||||
return self.predicate(value)
|
||||
else:
|
||||
return value == self.target_value
|
||||
|
||||
|
||||
class CompositeCondition(Condition):
|
||||
"""Allows combining conditions with AND/OR logic"""
|
||||
|
||||
def __init__(self, conditions: list[Condition], operator: str = "AND"):
|
||||
self.conditions = conditions
|
||||
self.operator = operator.upper()
|
||||
|
||||
def evaluate(self, context: HookContext) -> bool:
|
||||
if not self.conditions:
|
||||
return True
|
||||
|
||||
results = [condition.evaluate(context) for condition in self.conditions]
|
||||
|
||||
if self.operator == "AND":
|
||||
return all(results)
|
||||
elif self.operator == "OR":
|
||||
return any(results)
|
||||
else:
|
||||
raise ValueError(f"Unknown operator: {self.operator}")
|
||||
|
||||
|
||||
class Hook:
|
||||
"""Represents a complete hook with event, conditions, and executor"""
|
||||
|
||||
def __init__(self, event_type: EventType, conditions: list[Condition],
|
||||
executor: Callable, requires_modification: bool = False):
|
||||
self.event_type = event_type
|
||||
self.conditions = conditions
|
||||
self.executor = executor
|
||||
self.requires_modification = requires_modification
|
||||
|
||||
def matches(self, event_type: EventType, context: HookContext) -> bool:
|
||||
"""Checks if this hook should be executed for given event and context"""
|
||||
if self.event_type != event_type:
|
||||
return False
|
||||
|
||||
return all(condition.evaluate(context) for condition in self.conditions)
|
||||
|
||||
def execute(self, context: HookContext) -> Any:
|
||||
"""Executes the hook with given context"""
|
||||
return self.executor(context)
|
||||
|
||||
|
||||
class HookBuilder:
|
||||
"""Builder class for creating hooks with fluent interface"""
|
||||
|
||||
def __init__(self):
|
||||
self._event_type: Optional[EventType] = None
|
||||
self._conditions: list[Condition] = []
|
||||
self._executor: Optional[Callable] = None
|
||||
self._requires_modification: bool = False
|
||||
|
||||
# Event specification methods
|
||||
def on_render(self):
|
||||
"""Hook will be triggered on render event"""
|
||||
self._event_type = EventType.RENDER
|
||||
return self
|
||||
|
||||
def on_click(self):
|
||||
"""Hook will be triggered on click event"""
|
||||
self._event_type = EventType.CLICK
|
||||
return self
|
||||
|
||||
def on_hover(self):
|
||||
"""Hook will be triggered on hover event"""
|
||||
self._event_type = EventType.HOVER
|
||||
return self
|
||||
|
||||
def on_focus(self):
|
||||
"""Hook will be triggered on focus event"""
|
||||
self._event_type = EventType.FOCUS
|
||||
return self
|
||||
|
||||
# Condition methods
|
||||
def when_long_text(self, threshold: int = 100):
|
||||
"""Add condition: text length > threshold"""
|
||||
self._conditions.append(WhenLongText(threshold))
|
||||
return self
|
||||
|
||||
def when_editable(self, editable_paths: list[str] = None, editable_types: list[type] = None):
|
||||
"""Add condition: node is editable"""
|
||||
self._conditions.append(WhenEditable(editable_paths, editable_types))
|
||||
return self
|
||||
|
||||
def when_type(self, target_type: type):
|
||||
"""Add condition: node value is of specific type"""
|
||||
self._conditions.append(WhenType(target_type))
|
||||
return self
|
||||
|
||||
def when_key(self, key_pattern: Any):
|
||||
"""Add condition: node key matches pattern"""
|
||||
self._conditions.append(WhenKey(key_pattern))
|
||||
return self
|
||||
|
||||
def when_path(self, path_pattern: str):
|
||||
"""Add condition: JSON path matches pattern"""
|
||||
self._conditions.append(WhenPath(path_pattern))
|
||||
return self
|
||||
|
||||
def when_value(self, target_value: Any = None, predicate: Callable[[Any], bool] = None):
|
||||
"""Add condition: node value matches specific value or predicate"""
|
||||
self._conditions.append(WhenValue(target_value, predicate))
|
||||
return self
|
||||
|
||||
def when_custom(self, condition):
|
||||
"""Add custom condition (supports both Condition instances and predicate functions)."""
|
||||
if callable(condition) and not isinstance(condition, Condition):
|
||||
# Wrap the predicate function in a Condition class
|
||||
class PredicateCondition(Condition):
|
||||
def __init__(self, predicate):
|
||||
self.predicate = predicate
|
||||
|
||||
def evaluate(self, context):
|
||||
return self.predicate(context)
|
||||
|
||||
condition = PredicateCondition(condition) # Pass the function to the wrapper
|
||||
|
||||
elif not isinstance(condition, Condition):
|
||||
raise ValueError("when_custom expects a Condition instance or a callable predicate.")
|
||||
|
||||
self._conditions.append(condition)
|
||||
return self
|
||||
|
||||
def when_all(self, conditions: list[Condition]):
|
||||
"""Add composite condition with AND logic"""
|
||||
self._conditions.append(CompositeCondition(conditions, "AND"))
|
||||
return self
|
||||
|
||||
def when_any(self, conditions: list[Condition]):
|
||||
"""Add composite condition with OR logic"""
|
||||
self._conditions.append(CompositeCondition(conditions, "OR"))
|
||||
return self
|
||||
|
||||
# Modification flag
|
||||
def requires_modification(self):
|
||||
"""Indicates this hook will modify the state"""
|
||||
self._requires_modification = True
|
||||
return self
|
||||
|
||||
# Execution
|
||||
def execute(self, executor: Callable) -> Hook:
|
||||
"""Sets the executor function and builds the hook"""
|
||||
if not self._event_type:
|
||||
raise ValueError("Event type must be specified (use on_render(), on_click(), etc.)")
|
||||
|
||||
if not executor:
|
||||
raise ValueError("Executor function must be provided")
|
||||
|
||||
self._executor = executor
|
||||
|
||||
return Hook(
|
||||
event_type=self._event_type,
|
||||
conditions=self._conditions,
|
||||
executor=self._executor,
|
||||
requires_modification=self._requires_modification
|
||||
)
|
||||
|
||||
|
||||
class HookManager:
|
||||
"""Manages and executes hooks for JsonViewer"""
|
||||
|
||||
def __init__(self):
|
||||
self.hooks: list[Hook] = []
|
||||
|
||||
def add_hook(self, hook: Hook):
|
||||
"""Adds a hook to the manager"""
|
||||
self.hooks.append(hook)
|
||||
|
||||
def add_hooks(self, hooks: list[Hook]):
|
||||
"""Adds multiple hooks to the manager"""
|
||||
self.hooks.extend(hooks)
|
||||
|
||||
def find_matching_hooks(self, event_type: EventType, context: HookContext) -> list[Hook]:
|
||||
"""Finds all hooks that match the event and context"""
|
||||
return [hook for hook in self.hooks if hook.matches(event_type, context)]
|
||||
|
||||
def execute_hooks(self, event_type: EventType, context: HookContext) -> list[Any]:
|
||||
"""Executes all matching hooks and returns results"""
|
||||
matching_hooks = self.find_matching_hooks(event_type, context)
|
||||
results = []
|
||||
|
||||
for hook in matching_hooks:
|
||||
try:
|
||||
result = hook.execute(context)
|
||||
results.append(result)
|
||||
|
||||
# If this hook requires modification, we might want to stop here
|
||||
# or handle the modification differently
|
||||
if hook.requires_modification:
|
||||
# Could add callback to parent component here
|
||||
pass
|
||||
|
||||
except Exception as e:
|
||||
# Log error but continue with other hooks
|
||||
print(f"Hook execution error: {e}")
|
||||
continue
|
||||
|
||||
return results
|
||||
|
||||
def clear_hooks(self):
|
||||
"""Removes all hooks"""
|
||||
self.hooks.clear()
|
||||
|
||||
|
||||
# Example usage and factory functions
|
||||
def create_long_text_viewer_hook(threshold: int = 100) -> Hook:
|
||||
"""Factory function for common long text viewer hook"""
|
||||
|
||||
def text_viewer_component(context: HookContext):
|
||||
from fasthtml.components import Div, Span
|
||||
|
||||
value = context.get_value()
|
||||
truncated = value[:threshold] + "..."
|
||||
|
||||
return Div(
|
||||
Span(truncated, cls="text-truncated"),
|
||||
Span("Click to expand", cls="expand-hint"),
|
||||
cls="long-text-viewer"
|
||||
)
|
||||
|
||||
return (HookBuilder()
|
||||
.on_render()
|
||||
.when_long_text(threshold)
|
||||
.execute(text_viewer_component))
|
||||
|
||||
|
||||
def create_inline_editor_hook(editable_paths: list[str] = None) -> Hook:
|
||||
"""Factory function for common inline editor hook"""
|
||||
|
||||
def inline_editor_component(context: HookContext):
|
||||
from fasthtml.components import Input, Div
|
||||
|
||||
value = context.get_value()
|
||||
|
||||
return Div(
|
||||
Input(
|
||||
value=str(value),
|
||||
type="text" if isinstance(value, str) else "number",
|
||||
cls="inline-editor"
|
||||
),
|
||||
cls="editable-field"
|
||||
)
|
||||
|
||||
return (HookBuilder()
|
||||
.on_click()
|
||||
.when_editable(editable_paths)
|
||||
.requires_modification()
|
||||
.execute(inline_editor_component))
|
||||
Reference in New Issue
Block a user