融合 Clojure 声明式基因Clojure's Declarative Gene
与 Pythonic 全栈反应式Meets Pythonic Full-Stack Reactivity

class Counter(Component):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.count = signal(0)

    @server
    def increment(self, step: int = 1):
        self.count.set(self.count.get() + step)

    def render(self):
        count = self.count.get()
        return div({'class': 'card ...'},
            h2({}, f'Count: {count}'),
            button({'on_click': self.increment(1)}, '+1'),
        )

app = create_hiccl_app(HicclConfig(
    pages=menu(Counter),
))

class TwoClocks(Component):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._server_time = signal(time.time() * 1000)

    def mount(self):
        self._task = asyncio.create_task(
            self._tick_loop()
        )

    async def _tick_loop(self):
        while self._running:
            if self._session.transport.is_connected():
                self._server_time.set(time.time() * 1000)
            await asyncio.sleep(0.016)  # ~60fps

class ChatRoom(Component):
    topics = ["chat-messages"]

    @server
    def send_message(self, text: str, user: str = "Anonymous"):
        msg = {'user': user, 'text': text,
               'time': datetime.now().isoformat()}
        messages.append(msg)
        self.messages.set(list(messages))

    def on_broadcast(self, topic: str):
        if topic == "chat-messages":
            self.messages.set(list(messages))

class WebShellComponent(Component):
    async def on_stream_open(self, stream):
        self.pty = PtyProcess(detect_command())
        self.pty.start()
        self._tasks = [
            asyncio.create_task(self._pump_pty_to_stream()),
            asyncio.create_task(self._pump_stream_to_pty()),
            asyncio.create_task(self._watchdog()),
        ]

    async def _pump_pty_to_stream(self):
        data = await self._read_queue.get()
        await self.stream.send(data)

    async def _pump_stream_to_pty(self):
        async for data in self.stream:
            kind, value = parse_terminal_frame(bytes(data))
            if kind == 'resize':
                self.pty.resize(*value)
            else:
                self.pty.write(value)

class CryptoTrader(Component):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.trade_queue = Channel(maxsize=3)
        self.price_queue = Channel(maxsize=5)

    def mount(self):
        @go
        async def stream_orchestrator():
            selected, val = await alts_([
                self.price_queue, timeout(1500)
            ])
            if selected is self.price_queue:
                self._btc_price.set(val)

class CounterBox(Component):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.count = signal_with_history(
            10, max_snapshots=20
        )

    @server
    def step_up(self):
        self.count.set(self.count.get() + 1)

    @server
    def multiply_two(self):
        self.count.set(self.count.get() * 2)
    # Built-in TimeTravelPanel for undo/redo

# Flatten nested JSON into a queryable Database
db = Database.from_state({
    "name": "Volt Technologies",
    "employees": [
        {"id": "e1", "name": "Alice", "role": "Engineer"},
    ],
})

# Datalog-style declarative logic queries
results = db.solve(
    [Var("name"), Var("role")],
    [("?emp", "name", "?name"),
     ("?emp", "role", "?role")],
)

# Spec contracts guard @server method boundaries
UserSpec = spec.keys(req={
    "username": spec.string(min_len=3, max_len=20),
    "age": spec.integer(gte=18, lte=120),
})

@server(spec={"args": {"data": UserSpec}})
def update_profile(self, data: dict) -> bool:
    self.profile.set(data)
    return True
# SpecValidationError -> explain_data introspection