Streaming

IStream interface for building real-time data streams

Overview

The IStream interface enables building custom stream backends for real-time, multi-client data synchronization. It’s designed for applications like collaborative editing, live dashboards, and real-time chat.

IStream Interface

Abstract base class for stream implementations.

from abc import ABC, abstractmethod
from typing import Generic, TypeVar
from iii import IStream, StreamGetInput, StreamSetInput, StreamSetResult

TData = TypeVar("TData")

class IStream(ABC, Generic[TData]):
    @abstractmethod
    async def get(self, input: StreamGetInput) -> TData | None:
        """Get an item from the stream."""
        ...
    
    @abstractmethod
    async def set(self, input: StreamSetInput) -> StreamSetResult[TData] | None:
        """Set an item in the stream."""
        ...
    
    @abstractmethod
    async def delete(self, input: StreamDeleteInput) -> None:
        """Delete an item from the stream."""
        ...
    
    @abstractmethod
    async def list(self, input: StreamListInput) -> list[TData]:
        """Get all items in a group."""
        ...
    
    @abstractmethod
    async def list_groups(self, input: StreamListGroupsInput) -> list[str]:
        """List all groups in the stream."""
        ...
    
    @abstractmethod
    async def update(self, input: StreamUpdateInput) -> StreamSetResult[TData] | None:
        """Update an item in the stream with atomic operations."""
        ...

Registering a Stream

create_stream

Register stream functions for a given stream name:

from iii import IStream

class MyStream(IStream[dict]):
    async def get(self, input):
        # Implementation
        ...
    
    async def set(self, input):
        # Implementation
        ...
    
    # ... implement other methods

stream = MyStream()
iii.create_stream("my-stream", stream)

This automatically registers the following functions:

• stream::get(my-stream)
• stream::set(my-stream)
• stream::delete(my-stream)
• stream::list(my-stream)
• stream::list_groups(my-stream)
• stream::update(my-stream)

stream_name · str · required

Unique name for the stream

stream · IStream[Any] · required

Stream implementation instance

Input Types

StreamGetInput

Input for retrieving a single item:

class StreamGetInput(BaseModel):
    stream_name: str  # Stream identifier
    group_id: str     # Group/room identifier
    item_id: str      # Item identifier

StreamSetInput

Input for setting an item:

class StreamSetInput(BaseModel):
    stream_name: str  # Stream identifier
    group_id: str     # Group/room identifier
    item_id: str      # Item identifier
    data: Any         # Item data

StreamDeleteInput

Input for deleting an item:

class StreamDeleteInput(BaseModel):
    stream_name: str  # Stream identifier
    group_id: str     # Group/room identifier
    item_id: str      # Item identifier

StreamListInput

Input for listing all items in a group:

class StreamListInput(BaseModel):
    stream_name: str  # Stream identifier
    group_id: str     # Group/room identifier

StreamListGroupsInput

Input for listing all groups:

class StreamListGroupsInput(BaseModel):
    stream_name: str  # Stream identifier

StreamUpdateInput

Input for atomic updates:

class StreamUpdateInput(BaseModel):
    stream_name: str     # Stream identifier
    group_id: str        # Group/room identifier
    item_id: str         # Item identifier
    ops: list[UpdateOp]  # Update operations

Output Types

StreamSetResult

Result of set/update operations:

class StreamSetResult(BaseModel, Generic[TData]):
    old_value: TData | None  # Previous value
    new_value: TData | None  # New value

Update Operations

Atomic update operations for the update() method:

UpdateSet

Set a field to a value:

class UpdateSet(BaseModel):
    type: str = "set"
    path: str      # Field path (e.g., "name", "user.age")
    value: Any     # New value

UpdateIncrement

Increment a numeric field:

class UpdateIncrement(BaseModel):
    type: str = "increment"
    path: str         # Field path
    by: int | float   # Amount to increment

UpdateDecrement

Decrement a numeric field:

class UpdateDecrement(BaseModel):
    type: str = "decrement"
    path: str         # Field path
    by: int | float   # Amount to decrement

UpdateRemove

Remove a field:

class UpdateRemove(BaseModel):
    type: str = "remove"
    path: str  # Field path to remove

UpdateMerge

Merge an object:

class UpdateMerge(BaseModel):
    type: str = "merge"
    path: str   # Field path
    value: Any  # Object to merge

Example: In-Memory Stream

from typing import Any
from iii import (
    IStream,
    StreamGetInput,
    StreamSetInput,
    StreamSetResult,
    StreamDeleteInput,
    StreamListInput,
    StreamListGroupsInput,
    StreamUpdateInput,
)

class InMemoryStream(IStream[dict]):
    def __init__(self):
        self.data: dict[str, dict[str, dict]] = {}  # group_id -> item_id -> data
    
    async def get(self, input: StreamGetInput) -> dict | None:
        group = self.data.get(input.group_id, {})
        return group.get(input.item_id)
    
    async def set(self, input: StreamSetInput) -> StreamSetResult[dict] | None:
        if input.group_id not in self.data:
            self.data[input.group_id] = {}
        
        old_value = self.data[input.group_id].get(input.item_id)
        self.data[input.group_id][input.item_id] = input.data
        
        return StreamSetResult(
            old_value=old_value,
            new_value=input.data
        )
    
    async def delete(self, input: StreamDeleteInput) -> None:
        if input.group_id in self.data:
            self.data[input.group_id].pop(input.item_id, None)
    
    async def list(self, input: StreamListInput) -> list[dict]:
        group = self.data.get(input.group_id, {})
        return list(group.values())
    
    async def list_groups(self, input: StreamListGroupsInput) -> list[str]:
        return list(self.data.keys())
    
    async def update(self, input: StreamUpdateInput) -> StreamSetResult[dict] | None:
        if input.group_id not in self.data:
            self.data[input.group_id] = {}
        
        old_value = self.data[input.group_id].get(input.item_id, {})
        new_value = old_value.copy() if old_value else {}
        
        for op in input.ops:
            if op.type == "set":
                self._set_path(new_value, op.path, op.value)
            elif op.type == "increment":
                current = self._get_path(new_value, op.path) or 0
                self._set_path(new_value, op.path, current + op.by)
            elif op.type == "decrement":
                current = self._get_path(new_value, op.path) or 0
                self._set_path(new_value, op.path, current - op.by)
            elif op.type == "remove":
                self._remove_path(new_value, op.path)
            elif op.type == "merge":
                current = self._get_path(new_value, op.path) or {}
                if isinstance(current, dict) and isinstance(op.value, dict):
                    self._set_path(new_value, op.path, {**current, **op.value})
        
        self.data[input.group_id][input.item_id] = new_value
        
        return StreamSetResult(
            old_value=old_value,
            new_value=new_value
        )
    
    def _get_path(self, obj: dict, path: str) -> Any:
        keys = path.split(".")
        for key in keys:
            if isinstance(obj, dict):
                obj = obj.get(key)
            else:
                return None
        return obj
    
    def _set_path(self, obj: dict, path: str, value: Any) -> None:
        keys = path.split(".")
        for key in keys[:-1]:
            if key not in obj:
                obj[key] = {}
            obj = obj[key]
        obj[keys[-1]] = value
    
    def _remove_path(self, obj: dict, path: str) -> None:
        keys = path.split(".")
        for key in keys[:-1]:
            if key not in obj:
                return
            obj = obj[key]
        obj.pop(keys[-1], None)

# Register the stream
stream = InMemoryStream()
iii.create_stream("memory", stream)

Example: Redis-Backed Stream

import json
import redis.asyncio as redis
from iii import IStream, StreamGetInput, StreamSetInput, StreamSetResult

class RedisStream(IStream[dict]):
    def __init__(self, redis_url: str = "redis://localhost"):
        self.redis_url = redis_url
        self.client = None
    
    async def _get_client(self) -> redis.Redis:
        if self.client is None:
            self.client = await redis.from_url(self.redis_url)
        return self.client
    
    def _key(self, group_id: str, item_id: str) -> str:
        return f"stream:{group_id}:{item_id}"
    
    async def get(self, input: StreamGetInput) -> dict | None:
        client = await self._get_client()
        key = self._key(input.group_id, input.item_id)
        data = await client.get(key)
        return json.loads(data) if data else None
    
    async def set(self, input: StreamSetInput) -> StreamSetResult[dict] | None:
        client = await self._get_client()
        key = self._key(input.group_id, input.item_id)
        
        # Get old value
        old_data = await client.get(key)
        old_value = json.loads(old_data) if old_data else None
        
        # Set new value
        await client.set(key, json.dumps(input.data))
        
        return StreamSetResult(
            old_value=old_value,
            new_value=input.data
        )
    
    async def delete(self, input: StreamDeleteInput) -> None:
        client = await self._get_client()
        key = self._key(input.group_id, input.item_id)
        await client.delete(key)
    
    async def list(self, input: StreamListInput) -> list[dict]:
        client = await self._get_client()
        pattern = f"stream:{input.group_id}:*"
        keys = await client.keys(pattern)
        
        items = []
        for key in keys:
            data = await client.get(key)
            if data:
                items.append(json.loads(data))
        
        return items
    
    async def list_groups(self, input: StreamListGroupsInput) -> list[str]:
        client = await self._get_client()
        keys = await client.keys("stream:*")
        
        groups = set()
        for key in keys:
            parts = key.decode().split(":")
            if len(parts) >= 2:
                groups.add(parts[1])
        
        return list(groups)
    
    async def update(self, input: StreamUpdateInput) -> StreamSetResult[dict] | None:
        # Get current value
        current = await self.get(StreamGetInput(
            stream_name=input.stream_name,
            group_id=input.group_id,
            item_id=input.item_id
        )) or {}
        
        old_value = current.copy()
        
        # Apply operations
        for op in input.ops:
            if op.type == "set":
                self._set_path(current, op.path, op.value)
            elif op.type == "increment":
                value = self._get_path(current, op.path) or 0
                self._set_path(current, op.path, value + op.by)
            # ... implement other operations
        
        # Save updated value
        await self.set(StreamSetInput(
            stream_name=input.stream_name,
            group_id=input.group_id,
            item_id=input.item_id,
            data=current
        ))
        
        return StreamSetResult(
            old_value=old_value,
            new_value=current
        )
    
    def _get_path(self, obj: dict, path: str) -> Any:
        keys = path.split(".")
        for key in keys:
            obj = obj.get(key) if isinstance(obj, dict) else None
        return obj
    
    def _set_path(self, obj: dict, path: str, value: Any) -> None:
        keys = path.split(".")
        for key in keys[:-1]:
            if key not in obj:
                obj[key] = {}
            obj = obj[key]
        obj[keys[-1]] = value

# Register the stream
stream = RedisStream("redis://localhost:6379")
iii.create_stream("redis-stream", stream)

Example: Collaborative Todo List

import asyncio
from iii import III, IStream

iii = III("ws://localhost:49134")

# In-memory stream for todos
stream = InMemoryStream()
iii.create_stream("todos", stream)

# Client functions
async def add_todo(data):
    """Add a new todo item."""
    user_id = data["user_id"]
    todo = {
        "id": data["id"],
        "title": data["title"],
        "completed": False,
        "created_at": time.time()
    }
    
    await iii.call("stream::set(todos)", {
        "stream_name": "todos",
        "group_id": user_id,
        "item_id": todo["id"],
        "data": todo
    })
    
    return {"status": "created", "todo": todo}

async def complete_todo(data):
    """Mark a todo as completed."""
    user_id = data["user_id"]
    todo_id = data["todo_id"]
    
    result = await iii.call("stream::update(todos)", {
        "stream_name": "todos",
        "group_id": user_id,
        "item_id": todo_id,
        "ops": [
            {"type": "set", "path": "completed", "value": True}
        ]
    })
    
    return {"status": "updated", "todo": result["new_value"]}

async def list_todos(data):
    """List all todos for a user."""
    user_id = data["user_id"]
    
    todos = await iii.call("stream::list(todos)", {
        "stream_name": "todos",
        "group_id": user_id
    })
    
    return {"todos": todos}

iii.register_function("todos.add", add_todo)
iii.register_function("todos.complete", complete_todo)
iii.register_function("todos.list", list_todos)

async def main():
    await iii.connect()
    
    # Add todos
    await iii.call("todos.add", {
        "user_id": "user123",
        "id": "todo1",
        "title": "Buy groceries"
    })
    
    await iii.call("todos.add", {
        "user_id": "user123",
        "id": "todo2",
        "title": "Write documentation"
    })
    
    # List todos
    result = await iii.call("todos.list", {"user_id": "user123"})
    print(f"Todos: {result['todos']}")
    
    # Complete a todo
    await iii.call("todos.complete", {
        "user_id": "user123",
        "todo_id": "todo1"
    })
    
    # List again
    result = await iii.call("todos.list", {"user_id": "user123"})
    print(f"Updated todos: {result['todos']}")

if __name__ == "__main__":
    asyncio.run(main())

Use Cases

• Collaborative editing: Real-time document collaboration
• Live dashboards: Streaming metrics and analytics
• Chat applications: Multi-user chat rooms
• Gaming: Real-time game state synchronization
• IoT: Device state management
• Presence systems: Track online users