使用Pydantic和Streamlit回调实现持久化应用状态到JSON

本教程详细介绍了如何在Streamlit仪表板中实现应用状态的JSON持久化。我们将利用Pydantic定义结构化的应用状态模型，并通过其内置的序列化方法将状态高效地保存为JSON文件。文章还将展示如何结合Streamlit的`on_change`回调机制，在用户交互时自动触发状态保存，并提供从JSON文件加载状态的完整实现，确保仪表板刷新或重访时能无缝恢复之前的工作状态。

在开发交互式Streamlit仪表板时，用户对各种参数和行为的调整往往希望在刷新页面或下次访问时得到保留。这种需求催生了应用状态持久化的必要性。虽然Streamlit提供了st.session_state用于管理会话内的状态，但它通常不适用于跨会话或跨重启的持久化。将应用状态保存到外部文件（如JSON）是一种常见且有效的方法，能够确保用户体验的连续性和数据的一致性。

本教程将指导您如何结合Pydantic模型和Streamlit的事件回调机制，优雅地实现应用状态到JSON文件的持久化和加载。

Pydantic是一个强大的数据验证和设置管理库，非常适合定义结构化的应用状态。通过Pydantic模型，我们可以清晰地定义仪表板中所有需要持久化的参数及其数据类型。

考虑以下示例，它定义了一个包含相机选择、裁剪参数和处理流程配置的复杂应用状态：

import osimport jsonfrom typing import List, Optionalfrom pydantic import BaseModel, Field# 定义状态文件路径STATE_FILE_PATH = "application_state.json"class SelectCameraState(BaseModel):    """相机选择状态模型"""    selected_cameras: List[str] = Field(default_factory=list)class CropState(BaseModel):    """裁剪状态模型"""    crop_type: str = "Anchor"  # Anchor / Fixed    bbox: List[int] = Field(default_factory=lambda: [0, 0, 100, 100]) # x, y, width, height    anchor_class: str = "default_anchor"    anchor_position: List[int] = Field(default_factory=lambda: [50, 50]) # center_x, center_yclass ProcessState(BaseModel):    """处理流程状态模型"""    feature_extractor: str = "ResNet"    embedding_processor: str = "PCA"    outlier_detector: str = "IsolationForest"class ApplicationState(BaseModel):    """整个应用的综合状态模型"""    camera_select_state: SelectCameraState = Field(default_factory=SelectCameraState)    crop_state: CropState = Field(default_factory=CropState)    process_state: ProcessState = Field(default_factory=ProcessState) # 新增处理状态    class Config:        validate_assignment = True # 开启赋值验证

Pydantic模型设计要点：

Pydantic模型提供了便捷的方法来将实例序列化为JSON字符串，以及从JSON字符串反序列化回模型实例。

Pydantic v2+ 版本提供了 model_dump_json() 方法，可以方便地将模型实例转换为JSON字符串。对于Pydantic v1，您可以使用 json() 方法。

# 示例：序列化 ApplicationState 实例app_state_instance = ApplicationState()app_state_instance.camera_select_state.selected_cameras = ["camera_A", "camera_B"]app_state_instance.crop_state.crop_type = "Fixed"# 将模型实例序列化为美观的JSON字符串json_output = app_state_instance.model_dump_json(indent=2)print(json_output)

输出示例：

{  "camera_select_state": {    "selected_cameras": [      "camera_A",      "camera_B"    ]  },  "crop_state": {    "crop_type": "Fixed",    "bbox": [      0,      0,      100,      100    ],    "anchor_class": "default_anchor",    "anchor_position": [      50,      50    ]  },  "process_state": {    "feature_extractor": "ResNet",    "embedding_processor": "PCA",    "outlier_detector": "IsolationForest"  }}

要从JSON字符串或文件加载状态，可以使用Pydantic模型提供的类方法 model_validate_json() (Pydantic v2+) 或 parse_raw() (Pydantic v1)。

AI应用商店，提供即时交付、按需付费的人工智能应用服务

# 示例：从JSON字符串反序列化json_string_from_file = """{  "camera_select_state": {    "selected_cameras": [      "camera_X",      "camera_Y"    ]  },  "crop_state": {    "crop_type": "Anchor",    "bbox": [      10,      20,      200,      150    ],    "anchor_class": "custom_anchor",    "anchor_position": [      100,      75    ]  },  "process_state": {    "feature_extractor": "VGG",    "embedding_processor": "TSNE",    "outlier_detector": "DBSCAN"  }}"""# 使用 model_validate_json 从 JSON 字符串创建模型实例loaded_state = ApplicationState.model_validate_json(json_string_from_file)print(loaded_state.camera_select_state.selected_cameras) # 输出: ['camera_X', 'camera_Y']

为了方便管理，我们将状态的保存和加载逻辑封装成独立的函数。

def save_application_state(state: ApplicationState, file_path: str = STATE_FILE_PATH):    """    将应用状态保存到指定的JSON文件。    """    try:        with open(file_path, "w", encoding="utf-8") as f:            f.write(state.model_dump_json(indent=2))        # st.success(f"状态已保存到 {file_path}") # 在Streamlit应用中可用于反馈    except IOError as e:        # st.error(f"保存状态失败: {e}")        print(f"保存状态失败: {e}") # 命令行输出错误    except Exception as e:        # st.error(f"保存状态时发生未知错误: {e}")        print(f"保存状态时发生未知错误: {e}")def load_application_state(file_path: str = STATE_FILE_PATH) -> ApplicationState:    """    从指定的JSON文件加载应用状态。如果文件不存在或加载失败，则返回一个默认的ApplicationState实例。    """    if not os.path.exists(file_path):        print(f"状态文件 '{file_path}' 不存在，将创建默认状态。")        return ApplicationState()    try:        with open(file_path, "r", encoding="utf-8") as f:            json_data = f.read()        return ApplicationState.model_validate_json(json_data)    except json.JSONDecodeError as e:        print(f"加载状态失败: JSON解析错误 - {e}。将创建默认状态。")        return ApplicationState()    except Exception as e:        print(f"加载状态时发生未知错误: {e}。将创建默认状态。")        return ApplicationState()

注意事项：

Streamlit的on_change回调机制是实现用户交互时自动保存状态的关键。当一个组件的值发生变化时，on_change参数指定的函数会被调用。

我们将使用st.session_state来存储当前的ApplicationState实例，并在组件变化时更新它并触发保存。

import streamlit as stimport osimport jsonfrom typing import List, Optionalfrom pydantic import BaseModel, Field# ... (Pydantic模型定义和save_application_state, load_application_state函数定义同上) ...# 确保 STATE_FILE_PATH 已定义STATE_FILE_PATH = "application_state.json" def update_and_save_state(key: str, new_value):    """    更新st.session_state中的应用状态，并触发保存。    """    # 确保 session_state 中有 app_state    if 'app_state' not in st.session_state:        st.session_state.app_state = load_application_state()    # 根据 key 更新 app_state 的相应部分    # 这里需要根据实际的 Streamlit 控件和 Pydantic 模型结构进行映射    # 示例：如果 key 是 'selected_cameras'    if key == 'selected_cameras':        st.session_state.app_state.camera_select_state.selected_cameras = new_value    elif key == 'crop_type':        st.session_state.app_state.crop_state.crop_type = new_value    elif key == 'feature_extractor':        st.session_state.app_state.process_state.feature_extractor = new_value    # 可以添加更多 elif 来处理其他状态字段    save_application_state(st.session_state.app_state)    st.toast("状态已自动保存！") # 给出用户反馈# --- Streamlit 应用主逻辑 ---st.set_page_config(layout="wide", page_title="Streamlit 状态持久化示例")st.title("Streamlit 应用状态持久化到 JSON 示例")# 1. 初始化或加载应用状态if 'app_state' not in st.session_state:    st.session_state.app_state = load_application_state()    st.success("应用状态已加载。")current_state: ApplicationState = st.session_state.app_state# 2. 构建 Streamlit UI，并绑定 on_change 回调st.header("相机选择")selected_cameras_options = ["Camera_A", "Camera_B", "Camera_C", "Camera_D"]# 使用 st.multiselect 来选择相机selected_cameras = st.multiselect(    "选择要使用的相机:",    options=selected_cameras_options,    default=current_state.camera_select_state.selected_cameras,    key="camera_selector",    on_change=update_and_save_state,    args=('selected_cameras', st.session_state.camera_selector) # 传递 key 和新值)st.header("裁剪设置")crop_type_options = ["Anchor", "Fixed"]crop_type = st.radio(    "选择裁剪类型:",    options=crop_type_options,    index=crop_type_options.index(current_state.crop_state.crop_type),    key="crop_type_selector",    on_change=update_and_save_state,    args=('crop_type', st.session_state.crop_type_selector))st.header("处理流程配置")feature_extractor_options = ["ResNet", "VGG", "EfficientNet"]feature_extractor = st.selectbox(    "选择特征提取器:",    options=feature_extractor_options,    index=feature_extractor_options.index(current_state.process_state.feature_extractor),    key="feature_extractor_selector",    on_change=update_and_save_state,    args=('feature_extractor', st.session_state.feature_extractor_selector))# 3. 显示当前状态（调试用）st.subheader("当前应用状态 (JSON)")st.json(current_state.model_dump())st.markdown("---")st.write("请尝试修改上面的选项，然后刷新页面或关闭再打开应用，查看状态是否被保留。")

关键点说明：

通过Pydantic模型和Streamlit的on_change回调，我们实现了一个健壮且易于管理的应用状态持久化方案。

回顾与最佳实践：

通过遵循这些指导原则，您可以为您的Streamlit仪表板构建一个稳定、用户友好的持久化状态管理系统。

以上就是使用Pydantic和Streamlit回调实现持久化应用状态到JSON的详细内容，更多请关注乐哥常识网其它相关文章！