from __future__ import annotations from typing import Callable, Literal, Sequence from gradio.components import Component from gradio.context import Context, LocalContext from gradio.events import EventListener, EventListenerMethod from gradio.layouts import Column class Renderable: def __init__( self, fn: Callable, inputs: list[Component] | Component | None = None, triggers: EventListener | Sequence[EventListener] | None = None, concurrency_limit: int | None | Literal["default"] = "default", concurrency_id: str | None = None, ): if Context.root_block is None: raise ValueError("Reactive render must be inside a Blocks context.") self._id = len(Context.root_block.renderables) Context.root_block.renderables.append(self) self.column = Column() self.column_id = Column()._id self.fn = fn self.inputs = [inputs] if isinstance(inputs, Component) else inputs self.triggers: list[EventListenerMethod] = [] if isinstance(triggers, EventListener): triggers = [triggers] if triggers: self.triggers = [ EventListenerMethod( getattr(t, "__self__", None) if t.has_trigger else None, t.event_name, ) for t in triggers ] Context.root_block.set_event_trigger( self.triggers, self.apply, self.inputs, None, show_api=False, concurrency_limit=concurrency_limit, concurrency_id=concurrency_id, renderable=self, ) def apply(self, *args, **kwargs): column_copy = Column(render=False) column_copy._id = self.column_id LocalContext.is_render.set(True) LocalContext.render_block.set(column_copy) try: self.fn(*args, **kwargs) blocks_config = LocalContext.blocks_config.get() if blocks_config is None: raise ValueError("Reactive render must be inside a LocalContext.") blocks_config.blocks[self.column_id] = column_copy finally: LocalContext.is_render.set(False) LocalContext.render_block.set(None) def render( inputs: list[Component] | None = None, triggers: list[EventListener] | None = None, concurrency_limit: int | None | Literal["default"] = None, concurrency_id: str | None = None, ): def wrapper_function(fn): Renderable(fn, inputs, triggers, concurrency_limit, concurrency_id) return fn return wrapper_function