component_framework.core

 28class Component:
 29    """
 30    Base component class with lifecycle management.
 31
 32    Lifecycle:
 33        1. __init__(**params)
 34        2. mount() OR hydrate(state)
 35        3. handle_event(event, payload)
 36        4. before_render()
 37        5. render()
 38        6. dehydrate()
 39
 40    Slots:
 41        Components can declare named slots via the ``slots`` class variable.
 42        Child components are assigned to slots with ``fill_slot()`` and their
 43        rendered HTML is available in the template context under ``slots``.
 44    """
 45
 46    template_name: str | None = None
 47    renderer = None
 48    permission_classes: ClassVar[list[type["BasePermission"]]] = []
 49    slots: ClassVar[list[str]] = []
 50    """Slot names this component accepts. Empty list means *any* slot name is accepted."""
 51
 52    def __init__(self, **params):
 53        self.params = params
 54        self.state: dict[str, Any] = {}
 55        self.errors: dict[str, str] = {}
 56        self.id = params.get("component_id") or self._generate_id()
 57        self._mounted = False
 58        self._slot_components: dict[str, Component] = {}
 59
 60    # ---------- Lifecycle ----------
 61
 62    def _generate_id(self) -> str:
 63        """Generate unique component ID."""
 64        return f"component-{uuid4().hex[:8]}"
 65
 66    def mount(self):
 67        """Initialize component on first load. Override in subclasses."""
 68        self._mounted = True
 69
 70    def hydrate(self, state: dict):
 71        """Restore component from serialized state."""
 72        self.state.update(state)
 73        self._mounted = True
 74
 75    def dehydrate(self) -> dict:
 76        """Serialize component state for persistence."""
 77        return self.state.copy()
 78
 79    def before_render(self):
 80        """Called before rendering. Use for derived state computation."""
 81        pass
 82
 83    # ---------- Slots / Composition ----------
 84
 85    def fill_slot(self, slot_name: str, component: "Component") -> None:
 86        """
 87        Assign a child component to a named slot.
 88
 89        If the component declares specific ``slots``, *slot_name* must be one of
 90        them.  If ``slots`` is empty (the default), any name is accepted
 91        (permissive mode).
 92
 93        Args:
 94            slot_name: Target slot identifier.
 95            component: Child component instance to render in the slot.
 96
 97        Raises:
 98            ComponentError: If *slot_name* is not in the declared ``slots`` list.
 99        """
100        if self.slots and slot_name not in self.slots:
101            raise ComponentError(f"Unknown slot '{slot_name}'. Available: {self.slots}")
102        self._slot_components[slot_name] = component
103
104    def render_slots(self) -> dict[str, str]:
105        """
106        Render all filled slot components.
107
108        Returns:
109            Dict mapping slot names to their rendered HTML strings.
110        """
111        rendered: dict[str, str] = {}
112        for name, child in self._slot_components.items():
113            rendered[name] = child.render()
114        return rendered
115
116    # ---------- Optimistic UI ----------
117
118    def get_optimistic_patch(self, event: str, payload: dict) -> dict | None:
119        """
120        Return a partial state dict that the client can apply immediately before the server
121        responds, for the given event and payload.
122
123        This enables optimistic UI updates: the client shows the expected result right away
124        and reconciles with the actual server state once the response arrives. On error, the
125        client can roll back to the previous state.
126
127        Override in subclasses to enable optimistic updates for specific events. Return None
128        (the default) to disable optimistic updates for a given event.
129
130        Args:
131            event: The event name being dispatched (e.g., "increment").
132            payload: The event payload dict.
133
134        Returns:
135            A partial state dict to apply optimistically, or None to skip.
136
137        Example::
138
139            def get_optimistic_patch(self, event: str, payload: dict) -> dict | None:
140                if event == "increment":
141                    return {"count": self.state.get("count", 0) + payload.get("amount", 1)}
142                return None
143        """
144        return None
145
146    # ---------- Events ----------
147
148    def handle_event(self, event: str, payload: dict):
149        """
150        Route event to a **synchronous** handler method.
151
152        For async handlers (``async def on_*``), use :meth:`async_handle_event`
153        instead.  Calling this method with an async handler will raise
154        :class:`ComponentError`.
155
156        Args:
157            event: Event name (e.g., "increment")
158            payload: Event data
159
160        Raises:
161            EventNotFoundError: If handler not found
162            ComponentError: If handler raises exception or is async
163        """
164        handler = getattr(self, f"on_{event}", None)
165
166        if not handler:
167            raise EventNotFoundError(f"No handler for event: {event}")
168
169        if inspect.iscoroutinefunction(handler):
170            raise ComponentError(
171                f"Handler 'on_{event}' is async — use async_dispatch() or "
172                "async_handle_event() instead of the sync variants."
173            )
174
175        try:
176            handler(**payload)
177        except TypeError as e:
178            raise ComponentError(f"Invalid payload for {event}: {e}") from e
179        except Exception as e:
180            logger.exception(f"Error handling {event} in {self.__class__.__name__}")
181            raise ComponentError(f"Error handling {event}") from e
182
183    async def async_handle_event(self, event: str, payload: dict):
184        """
185        Route event to handler, awaiting if the handler is async.
186
187        Works with both ``def on_*`` and ``async def on_*`` handlers.
188
189        Args:
190            event: Event name (e.g., "increment")
191            payload: Event data
192
193        Raises:
194            EventNotFoundError: If handler not found
195            ComponentError: If handler raises exception
196        """
197        handler = getattr(self, f"on_{event}", None)
198
199        if not handler:
200            raise EventNotFoundError(f"No handler for event: {event}")
201
202        try:
203            result = handler(**payload)
204            if inspect.isawaitable(result):
205                await result
206        except TypeError as e:
207            raise ComponentError(f"Invalid payload for {event}: {e}") from e
208        except Exception as e:
209            logger.exception(f"Error handling {event} in {self.__class__.__name__}")
210            raise ComponentError(f"Error handling {event}") from e
211
212    # ---------- Rendering ----------
213
214    def get_context(self) -> dict:
215        """
216        Build template context. Does not expose full component.
217
218        Override to add custom context variables.
219        The returned dict always includes a ``slots`` key containing the
220        rendered HTML of any filled child components.
221        """
222        return {
223            "state": self.state,
224            "errors": self.errors,
225            "component_id": self.id,
226            "slots": self.render_slots(),
227        }
228
229    def render(self) -> str:
230        """Render component to HTML."""
231        if not self.renderer:
232            raise ComponentError("No renderer configured")
233
234        if not self.template_name:
235            raise ComponentError("No template_name specified")
236
237        self.before_render()
238
239        return self.renderer.render(
240            self.template_name,
241            self.get_context(),
242        )
243
244    # ---------- Dispatch ----------
245
246    def dispatch(
247        self,
248        event: str | None = None,
249        payload: dict | None = None,
250        state: dict | None = None,
251    ) -> dict:
252        """
253        Synchronous entry point for component execution.
254
255        For components with ``async def on_*`` handlers, use
256        :meth:`async_dispatch` instead.
257
258        Args:
259            event: Event name to handle
260            payload: Event data
261            state: Serialized state to restore
262
263        Returns:
264            Dict with 'html' and 'state' keys
265        """
266        try:
267            # Lifecycle: mount or hydrate
268            if state:
269                self.hydrate(state)
270            else:
271                self.mount()
272
273            # Handle event if provided
274            if event:
275                self.handle_event(event, payload or {})
276
277            # Render
278            html = self.render()
279
280            return {
281                "html": html,
282                "state": self.dehydrate(),
283                "component_id": self.id,
284                "slots": self.render_slots(),
285            }
286
287        except Exception:
288            logger.exception(f"Error in {self.__class__.__name__}.dispatch()")
289            raise
290
291    async def async_dispatch(
292        self,
293        event: str | None = None,
294        payload: dict | None = None,
295        state: dict | None = None,
296    ) -> dict:
297        """
298        Async entry point for component execution.
299
300        Works with both sync and async event handlers.  Use this from async
301        adapters (FastAPI, Litestar, WebSocket) to support ``async def on_*``
302        handlers.
303
304        Args:
305            event: Event name to handle
306            payload: Event data
307            state: Serialized state to restore
308
309        Returns:
310            Dict with 'html' and 'state' keys
311        """
312        try:
313            # Lifecycle: mount or hydrate
314            if state:
315                self.hydrate(state)
316            else:
317                self.mount()
318
319            # Handle event if provided
320            if event:
321                await self.async_handle_event(event, payload or {})
322
323            # Render
324            html = self.render()
325
326            return {
327                "html": html,
328                "state": self.dehydrate(),
329                "component_id": self.id,
330                "slots": self.render_slots(),
331            }
332
333        except Exception:
334            logger.exception(f"Error in {self.__class__.__name__}.async_dispatch()")
335            raise

16class ComponentError(Exception):
17    """Base exception for component errors."""
18
19    pass

22class EventNotFoundError(ComponentError):
23    """Raised when an event handler is not found."""
24
25    pass

338class StateSerializer:
339    """Handles safe serialization/deserialization of component state.
340
341    Class attributes:
342        warn_bytes: Emit a warning when serialised state exceeds this size
343            (default 64 KB).  Set to ``0`` to disable warnings.
344        max_bytes: Raise :class:`ComponentError` when serialised state exceeds
345            this size (default 512 KB).  Set to ``0`` to disable the hard limit.
346    """
347
348    warn_bytes: int = 64 * 1024  # 64 KB
349    max_bytes: int = 512 * 1024  # 512 KB
350
351    @staticmethod
352    def serialize(state: dict) -> str:
353        """Serialize state to JSON string.
354
355        Emits a warning if the result exceeds :attr:`warn_bytes` and raises
356        :class:`ComponentError` if it exceeds :attr:`max_bytes`.
357        """
358        serialized = json.dumps(state, default=str)
359        size = len(serialized)
360
361        if StateSerializer.max_bytes and size > StateSerializer.max_bytes:
362            raise ComponentError(
363                f"Component state is {size:,} bytes "
364                f"(hard limit: {StateSerializer.max_bytes:,}). "
365                "Move large data out of state — store IDs/keys instead of "
366                "full objects, or use server-side caching."
367            )
368
369        if StateSerializer.warn_bytes and size > StateSerializer.warn_bytes:
370            logger.warning(
371                "Component state is %s bytes (threshold: %s). "
372                "Consider moving large data out of state.",
373                f"{size:,}",
374                f"{StateSerializer.warn_bytes:,}",
375            )
376
377        return serialized
378
379    @staticmethod
380    def deserialize(data: str) -> dict:
381        """Deserialize state from JSON string."""
382        if not data:
383            return {}
384        return json.loads(data)

11def compose(
12    parent_cls: type[Component],
13    *,
14    params: dict[str, Any] | None = None,
15    **slot_components: Component,
16) -> Component:
17    """
18    Convenience function: instantiate a parent component and fill its slots in one call.
19
20    Keyword arguments whose values are ``Component`` instances are treated as
21    slot assignments; everything else is forwarded to the parent constructor
22    via *params*.
23
24    Args:
25        parent_cls: The component class to instantiate.
26        params: Explicit keyword arguments forwarded to the parent constructor.
27        **slot_components: Mapping of slot names to child ``Component`` instances.
28
29    Returns:
30        The fully-assembled parent component (not yet dispatched).
31
32    Raises:
33        ComponentError: If a slot name is not accepted by the parent.
34
35    Example::
36
37        card = compose(
38            CardComponent,
39            params={"title": "My Card"},
40            header=HeaderComponent(text="Title"),
41            body=BodyComponent(items=[1, 2, 3]),
42        )
43        result = card.dispatch()
44    """
45    parent = parent_cls(**(params or {}))
46    for slot_name, child in slot_components.items():
47        parent.fill_slot(slot_name, child)
48    return parent

 51class SlotRenderer:
 52    """
 53    Mixin for renderers that support slot injection into templates.
 54
 55    Subclass your concrete ``Renderer`` together with this mixin to gain a
 56    ``render_with_slots`` helper that merges pre-rendered slot HTML into the
 57    template context before rendering.
 58
 59    Example::
 60
 61        class MyRenderer(SlotRenderer, Renderer):
 62            def render(self, template_name, context):
 63                ...
 64
 65            def render_with_slots(self, template_name, context, slots):
 66                ctx = self.merge_slot_context(context, slots)
 67                return self.render(template_name, ctx)
 68    """
 69
 70    @staticmethod
 71    def merge_slot_context(
 72        context: dict[str, Any],
 73        slots: dict[str, str],
 74    ) -> dict[str, Any]:
 75        """
 76        Return a *new* context dict with ``slots`` merged in.
 77
 78        If the context already contains a ``slots`` key it will be overwritten
 79        with the provided *slots* mapping.
 80
 81        Args:
 82            context: Original template context.
 83            slots: Mapping of slot names to rendered HTML.
 84
 85        Returns:
 86            A new dict combining *context* and the slot HTML.
 87        """
 88        merged = {**context, "slots": slots}
 89        return merged
 90
 91    def render_with_slots(
 92        self,
 93        template_name: str,
 94        context: dict[str, Any],
 95        slots: dict[str, str],
 96    ) -> str:
 97        """
 98        Render a template with slot HTML injected into the context.
 99
100        The default implementation merges the slots into the context and
101        delegates to ``self.render()``.  Override for custom behaviour.
102
103        Args:
104            template_name: Name/path of the template.
105            context: Template variables.
106            slots: Pre-rendered slot HTML keyed by slot name.
107
108        Returns:
109            Rendered HTML string.
110        """
111        merged = self.merge_slot_context(context, slots)
112        # Delegate to the concrete Renderer.render() provided by the subclass.
113        return self.render(template_name, merged)  # type: ignore[attr-defined]

 11class FormComponent(Component):
 12    """
 13    Form component with Pydantic validation.
 14
 15    Usage:
 16        class UserFormSchema(BaseModel):
 17            name: str
 18            email: EmailStr
 19            age: int = Field(ge=0, le=120)
 20
 21        @registry.register("user_form")
 22        class UserForm(FormComponent):
 23            schema = UserFormSchema
 24            template_name = "user_form"
 25
 26            def on_submit(self):
 27                # Validation happens automatically
 28                # Access validated data in self.validated_data
 29                pass
 30    """
 31
 32    # Pydantic model for validation
 33    schema: ClassVar[type[BaseModel] | None] = None
 34
 35    def __init__(self, **params):
 36        super().__init__(**params)
 37        self.validated_data: dict[str, Any] | None = None
 38        self.field_errors: dict[str, list[str]] = {}
 39
 40    # ---------- Validation ----------
 41
 42    def validate(self, data: dict) -> bool:
 43        """
 44        Validate data against schema.
 45
 46        Args:
 47            data: Form data to validate
 48
 49        Returns:
 50            True if valid, False otherwise
 51        """
 52        if not self.schema:
 53            # No schema, skip validation
 54            self.validated_data = data
 55            return True
 56
 57        try:
 58            validated = self.schema(**data)
 59            self.validated_data = validated.model_dump()
 60            self.errors = {}
 61            self.field_errors = {}
 62            return True
 63        except ValidationError as e:
 64            self.validated_data = None
 65            self.errors["form"] = "Validation failed"
 66
 67            # Convert Pydantic errors to field errors
 68            self.field_errors = {}
 69            for error in e.errors():
 70                field = str(error["loc"][0]) if error["loc"] else "form"
 71                message = error["msg"]
 72
 73                if field not in self.field_errors:
 74                    self.field_errors[field] = []
 75                self.field_errors[field].append(message)
 76
 77            return False
 78
 79    def get_field_errors(self, field_name: str) -> list[str]:
 80        """Get validation errors for a specific field."""
 81        return self.field_errors.get(field_name, [])
 82
 83    def has_field_error(self, field_name: str) -> bool:
 84        """Check if field has validation errors."""
 85        return field_name in self.field_errors
 86
 87    # ---------- Form State ----------
 88
 89    def mount(self):
 90        """Initialize form state."""
 91        super().mount()
 92        self.state["form_data"] = {}
 93        self.state["submitted"] = False
 94        self.state["is_valid"] = False
 95
 96    def get_field_value(self, field_name: str, default: Any = "") -> Any:
 97        """Get current value of form field."""
 98        return self.state.get("form_data", {}).get(field_name, default)
 99
100    def set_field_value(self, field_name: str, value: Any):
101        """Set value of form field."""
102        if "form_data" not in self.state:
103            self.state["form_data"] = {}
104        self.state["form_data"][field_name] = value
105
106    # ---------- Event Handlers ----------
107
108    def on_submit(self):
109        """
110        Handle form submission.
111
112        Override in subclasses to add custom logic.
113        Validation happens automatically before this is called.
114        """
115        pass
116
117    def on_field_change(self, field: str, value: Any):
118        """Handle field value change."""
119        self.set_field_value(field, value)
120
121        # Clear field error on change
122        if field in self.field_errors:
123            del self.field_errors[field]
124
125    def handle_event(self, event: str, payload: dict):
126        """Handle events with automatic validation for submit."""
127        if event == "submit":
128            # Extract form data
129            form_data = payload.get("form_data", self.state.get("form_data", {}))
130            self.state["form_data"] = form_data
131            self.state["submitted"] = True
132
133            # Validate
134            is_valid = self.validate(form_data)
135            self.state["is_valid"] = is_valid
136
137            if is_valid:
138                # Call submit handler
139                self.on_submit()
140        else:
141            # Regular event handling
142            super().handle_event(event, payload)
143
144    # ---------- Context ----------
145
146    def get_context(self) -> dict:
147        """Add form-specific context."""
148        context = super().get_context()
149        context.update(
150            {
151                "form_data": self.state.get("form_data", {}),
152                "field_errors": self.field_errors,
153                "is_valid": self.state.get("is_valid", False),
154                "submitted": self.state.get("submitted", False),
155            }
156        )
157        return context