Schema declarations on the Application state allows Framework to handle complex serialisation scenarios, while also allowing your IDE and toolchains to provide autocomplete and type checking.

​
Schema declaration

import writer as wf

class AppSchema(wf.StreamsyncState):
    counter: int

initial_state = wf.init_state({
    "counter": 0
}, schema=AppSchema)

# Event handler
# It receives the session state as an argument and mutates it
def increment(state: AppSchema):
    state.counter += 1

Accessing an attribute by its key is always possible.

def increment(state: AppSchema):
    state["counter"] += 1

Attributes missing from the schema remain accessible by their key.

initial_state = wf.init_state({
    "counter": 0,
    "message": None
}, schema=AppSchema)

def increment(state: AppSchema):
    state['message'] = "Hello pigeon"

​
Schema composition

Schema composition allows you to model a complex Application state.

class MyappSchema(wf.State):
    title: str

class AppSchema(wf.StreamsyncState):
    my_app: MyappSchema
    counter: int

initial_state = wf.init_state({
    "counter": 0,
    "my_app": {
        "title": "Nested value"
    }
}, schema=AppSchema)

​
Multi-level dictionary

Some components like Vega Lite Chart require specifying a graph in the form of a multi-level dictionary.

A schema allows you to specify that an attribute which contains a dictionary must be treated as a dictionary, rather than as a group of state.

class AppSchema(wf.StreamsyncState):
    vegas_graph: dict

# Without schema, this handler is execute only once
def handle_vega_graph(state: AppSchema):
    graph = state.vega_graph
    graph["data"]["values"][0]["b"] += 1000
    state.vega_graph = graph
    
initial_state = wf.init_state({
    "vegas_graph": {
        "data": {
            "values": [
                {"a": "C", "b": 2}, {"a": "C", "b": 7}, {"a": "C", "b": 4},
                {"a": "D", "b": 1}, {"a": "D", "b": 2}, {"a": "D", "b": 6},
                {"a": "E", "b": 8}, {"a": "E", "b": 4}, {"a": "E", "b": 7}
            ]
        },
        "mark": "bar",
        "encoding": {
            "x": {"field": "a", "type": "nominal"},
            "y": {"aggregate": "average", "field": "b", "type": "quantitative"}
        }
    },
}, schema=AppSchema)

​
Type checking

A schema allows you to check the integrity of your back-end using the type system. The code below will raise an error with mypy.

$ mypy apps/myapp/main.py
apps/myapp/main.py:7: error: "AppSchema" has no attribute "countr"; maybe "counter"?  [attr-defined]

Here is the code, can you spot the error ?

import writer as wf

class AppSchema(wf.StreamsyncState):
    counter: int

def increment(state: AppSchema):
    state.countr += 1

initial_state = wf.init_state({
    "counter": 26,
}, schema=AppSchema)
SessionsStylesheets