Full example of a global state

This is a follow-up to Simultaneous multipages (new API), widget persistence, duplicate widgets, multiprocessing, robust data editors - :speech_balloon: Show the Community! - Streamlit but implementing a global state as recommended by Asaurus1 here for the use case of running Streamlit on an unstable platform.

This example incorporates the following features:

  • New multipage functionality
  • Widget persistence between pages
  • Multiprocessing
  • Robust data editors (e.g., content is preserved upon switching pages… you can just delete these objects if you don’t care about this)
  • The state of everything is saved using a global state so that you can e.g. hit refresh or use an unstable deployment of Streamlit that will not start anew but will rather continually retain the complete state of your app.

The repository containing these files is here and is below except for this optional file:

# app.py

# Import relevant libraries
import streamlit as st
import one
import two
import three
import global_state_lib as gsl

# Main function
def main():

    # Load the global state
    global_state = gsl.get_global_state()

    # Use the new st.naviation()/st.Page() API to create a multi-page app
    pg = st.navigation({
        'first section':
            [st.Page(one.main, title="Home", url_path='home'),
             st.Page(two.main, title="Second page", url_path='two')],
        'second section':
            [st.Page(three.main, title="Third page", url_path='three')],
        })

    # This is needed for the st.dataframe_editor() class (https://github.com/andrew-weisman/streamlit-dataframe-editor) but is also useful for seeing where we are and where we've been
    # This is necessary so we know to "fast forward" the editable dataframe when we return to a page with one on it
    global_state['current_page_name'] = pg.url_path if pg.url_path != '' else 'Home'
    if 'previous_page_name' not in global_state:
        global_state['previous_page_name'] = global_state['current_page_name']

    # On every page, display its title
    st.title(pg.title)

    # Output where we are and where we just were
    st.write(f'Your page location: {global_state["current_page_name"]}')
    st.write(f'Previous page location: {global_state["previous_page_name"]}')

    # Render the select page
    pg.run()

    # Update the previous page location
    global_state['previous_page_name'] = global_state['current_page_name']

# Needed for rendering pages that use multiprocessing (https://docs.python.org/3/library/multiprocessing.html#the-spawn-and-forkserver-start-methods)
if __name__ == '__main__':
    main()
# one.py

# Import relevant libraries
import streamlit as st
import global_state_lib as gsl

# Main function
def main():

    # Load the global state
    global_state = gsl.get_global_state()

    # Slider widget
    key = 'slider'
    if key not in global_state:
        global_state[key] = 40
    st.slider("Slider", 0, 100, key=key, value=global_state[key], on_change=gsl.assign_to_global_state, args=(global_state, key))

    # Display dataframe_editor2's contents (from three.py)
    if 'dataframe_editor2' in global_state:
        st.dataframe(global_state['dataframe_editor2'].reconstruct_edited_dataframe(), hide_index=True)
    else:
        st.write('No dataframe editors have been initialized; do so on the third page')
# two.py

# Import relevant libraries
import streamlit as st
import multiprocessing as mp
import global_state_lib as gsl

# Function definition for testing multiprocessing
def f(x):
    return x*x

# Main function
def main():

    # Load the global state
    global_state = gsl.get_global_state()
    
    # Simulate cleanly returning from a page
    if st.button('Return'):
        st.warning('Returning')
        st.button('Restore page')
        return
    
    # Toggle widget
    key = 'toggle'
    if key not in global_state:
        global_state[key] = False
    st.toggle('Toggle me', key=key, value=global_state[key], on_change=gsl.assign_to_global_state, args=(global_state, key))

    # Run a parallel process
    if st.button('Run parallel process'):
        with mp.get_context('forkserver').Pool(4) as p:
            st.write(p.map(f, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]))
# three.py

# Import relevant libraries
import streamlit as st
import multiprocessing as mp
import streamlit_dataframe_editor_global as sde
import pandas as pd
import global_state_lib as gsl

# Function definition for testing multiprocessing
def f(x):
    return x**3

# Main function
def main():

    # Load the global state
    global_state = gsl.get_global_state()

    # Selectbox widget
    key = 'selectbox'
    if key not in global_state:
        global_state[key] = 'one'
    st.selectbox('Selectbox', ['one', 'two', 'three'], key=key, index=['one', 'two', 'three'].index(global_state[key]), on_change=gsl.assign_to_global_state, args=(global_state, key))

    # Run a parallel process
    if st.button('Run parallel process'):
        with mp.get_context('forkserver').Pool(4) as p:
            st.write(p.map(f, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]))

    # Create a default dataframe
    df_default = pd.DataFrame({'a': [1, 2, 3], 'b': [False, False, True], 'c': [4, 5, 6]})

    # Dataframe editor widget 1
    key = 'dataframe_editor1'
    if key not in global_state:
        global_state[key] = sde.DataframeEditor(df_orig=df_default, df_description='first_df', state_holder=global_state)
    global_state[key].render_data_editor(on_change=gsl.assign_to_global_state, num_rows='dynamic')

    # Dataframe editor widget 2
    key = 'dataframe_editor2'
    if key not in global_state:
        global_state[key] = sde.DataframeEditor(df_orig=df_default, df_description='second_df', state_holder=global_state)
    global_state[key].render_data_editor(on_change=gsl.assign_to_global_state, num_rows='dynamic')
# global_state_lib.py

# This is a library that contains the GlobalState class and functions to manage the global state in Streamlit. Using this library will continuously update an app's state so e.g. on unstable platforms if the session is temporarily disconnected you won't be left with an empty session state.

# Import relevant libraries
import streamlit as st
import streamlit_dataframe_editor_global as sde


# Define a class to hold the global state
class GlobalState:

    # Constructor
    def __init__(self):
        pass

    # Method to return all attribute names in the object
    def get_key_names(self):
        return list(self.__dict__.keys())

    # Method to reset the global state. We can safely assume that defaults are defined in the main code, just as we would with Streamlit session state
    def reset_global_state(self):
        for attr in self.get_key_names():  # Create a list to safely iterate
            delattr(self, attr)

    # Implement __setitem__ to support item assignment
    def __setitem__(self, key, value):
        setattr(self, key, value)

    # Optionally, implement __getitem__ to support item access
    def __getitem__(self, key):
        return getattr(self, key, None)
    
    # Implement __contains__ to support 'in' checks
    def __contains__(self, key):
        return hasattr(self, key)


# Need this function to assign to the global state via the session state while avoiding the Streamlit "every other" issue
def assign_to_global_state(global_state, common_key, callback=None, args=(), kwargs=None):
    if kwargs is None:
        kwargs = {}
    global_state[common_key] = st.session_state[common_key]
    del st.session_state[common_key]  # don't need this anymore and in fact if we didn't delete it, we'd likely have problems since widgets would then be set by both the initial value and the session state key
    if callback is not None:
        callback(*args, **kwargs)


# Function to create or retrieve the global state instance
@st.cache_resource
def initialize_global_state():
    return GlobalState()


# Get the global state and fast forward the editable dataframes if necessary
def get_global_state():
    global_state = initialize_global_state()
    if 'fresh_session_state' not in st.session_state:
        for key in global_state.get_key_names():
            if key.endswith('__dataframe_editor'):
                key_changes_dict = key
                key_df_orig = key.removesuffix('_changes_dict' + '__dataframe_editor')
                sde.fast_forward_editable_dataframe_in_state(global_state, key_df_orig, key_changes_dict)
        st.session_state['fresh_session_state'] = False
    return global_state

This topic was automatically closed 180 days after the last reply. New replies are no longer allowed.