reactpie 0.1.1

Python reactive state management framework with server/client incremental sync

ReactPie

Reactive state management for Python with server/client incremental sync.

English | 中文

---

Introduction

ReactPie makes server-side data mutations behave like slicing a pie — each attribute change or container operation produces a precise incremental delta, serialized into a compact binary protocol and pushed over WebSocket in real time. Clients automatically reconstruct a fully consistent reactive mirror of the game state.

Zero external dependencies. Pure Python standard library.

Features

• Field-level deltas — generated at mutation time, no post-hoc diffing
• Reactive containers — RList / RDict / RSet wrap native containers, mutations sync automatically
• Binary protocol — custom TAG + zigzag + LEB128 encoding, more compact than JSON
• Lifecycle management — reference-counted connect / disconnect / die
• Declarative callbacks — watch() for field changes, on_create / on_delete for entity lifecycle

Installation

pip install reactpie

Development mode:

git clone https://github.com/Darksky/ai-game-arena
cd ai-game-arena/python-package/reactpie
pip install -e ".[dev]"

Quick Start

import asyncio
from reactpie import ReactServer, ReactClient, RClass, reactive, watch

# ── Define a data model ──

@router
class User(RClass):
    name = reactive(str, default="")
    hp = reactive(int, default=100)

    def __rinit__(self, name: str = ""):
        self.name = name

# ── Server ──

server = ReactServer()
server.include_type(User)
server.append(router)

client = ReactClient()
client.include_type(User)

# ── Client-side listeners ──

@client.on_create(User)
def on_create(instance: User):
    print(f"Player created: {instance.name}")

    @watch(instance, "hp")
    def on_hp_change(inst, delta):
        print(f"{inst.name} HP: {inst.hp}")

# ── Run ──

async def main():
    async def runner():
        async for packet in server.serve():
            client.push(packet)

    task = asyncio.create_task(runner())

    user = User("Alice")
    server.collect(user)
    user.hp -= 20    # client automatically receives the HP delta

    await asyncio.sleep(1)
    task.cancel()
    await server.wait_stop()

asyncio.run(main())

Modules

Module	Responsibility
reactpie.core	Reactive base classes, serialization protocol, watch(), context & routing
reactpie.server	ReactServer — instant/batch push modes, serve() async generator
reactpie.client	ReactClient — deserialization, state mirroring, lifecycle callbacks

API Overview

RClass — Reactive Data Model

from reactpie import RClass, reactive, reactive_self

class Player(RClass):
    hp = reactive(int, default=100)
    name = reactive(str, default="")
    inventory = reactive(list, default=[])   # auto-wrapped as RList
    target = reactive_self()                 # self-referencing field

    def __rinit__(self, name: str = ""):     # replaces __init__
        self.name = name

    def __post_load__(self):                 # called after client-side reconstruction
        ...

watch — Change Listeners

from reactpie import watch, DeltaType

watch(obj, callback=cb)                          # listen to all deltas
watch(obj, callback=cb, catch=DeltaType.UPDATE)  # filter by delta type
watch(obj, callback=cb, field="hp")              # filter by field name
watch(obj, callback=cb, depth=0)                 # filter by depth

@watch(obj, field="x")                           # decorator form
def on_x_change(instance, delta):
    ...

ReactServer / ReactClient

# Server
server = ReactServer()
server.include_type(MyModel)
server.collect(root_instance)             # mark as root, triggers client on_create

async for packet in server.serve():       # yields MultiDelta bytes
    await websocket.send(packet)

# Client
client = ReactClient()
client.include_type(MyModel)

@client.on_create(MyModel)               # fires when a ROOT delta arrives
def handle(instance): ...

@client.on_delete(MyModel)               # fires when refcount drops to zero
def handle(instance): ...

client.push(raw_bytes)                    # deserialize and apply

Reactive Containers

from reactpie import RList, RDict, RSet

party = RList()
party.append(alice)     # → reactive_connect, live count +1
party.remove(alice)     # → reactive_disconnect, live count -1
party.clear()           # → disconnect all elements

# RDict, RSet work the same way

Binary Protocol

MultiDelta packet format:
┌──────────┬──────────┬───────────┬─────────────────────┐
│ Magic (2)│ Ver  (1) │ Count (2) │  Delta × Count      │
│ "RD"     │  0x01    │ u16 LE    │                     │
└──────────┴──────────┴───────────┴─────────────────────┘

DeltaType	Description
UPDATE	Field assignment
ROOT / CREATE	Instance creation (ROOT triggers on_create, CREATE registers only)
APPEND / EXTEND / INSERT	List append operations
REMOVE / POP / CLEAR	Container removal
SET	Index/key assignment

Examples

Example	Description
01_simple_e2e.py	Single-file end-to-end via asyncio.Queue
game_schema.py	Shared data models (Player, Projectile, GameState)
game_server.py	WebSocket game server
game_client.py	WebSocket game client
run_game.py	Single-process launcher

python examples/01_simple_e2e.py    # single-file demo
python examples/run_game.py         # multi-file WebSocket demo

Testing

python -m pytest tests/ -v       # 242 tests

Project Structure

reactpie/
├── reactpie/
│   ├── __init__.py    # public API exports
│   ├── core.py        # framework core
│   ├── server.py      # ReactServer
│   └── client.py      # ReactClient
├── examples/          # example code
├── tests/             # 242 unit tests
├── docs/              # logo & docs
├── pyproject.toml
└── README.md

License

MIT