使用Pydantic和Streamlit回调实现持久化应用状态到JSON-Python教程-PHP中文网

使用pydantic和streamlit回调实现持久化应用状态到json

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

引言：Streamlit应用状态持久化的重要性

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

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

使用Pydantic设计应用状态模型

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

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

import os
import json
from typing import List, Optional
from 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_y

class 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模型设计要点：

清晰的结构： 将相关参数分组到独立的子模型中，提高可读性和模块化。
默认值： 为所有字段设置合适的默认值，确保首次加载或文件不存在时应用能正常启动。
Field(default_factory=...)： 对于可变类型（如列表、字典），使用default_factory来提供默认值，避免所有实例共享同一个可变对象。
validate_assignment = True： 确保每次对模型实例的赋值操作都会进行验证。

Pydantic模型的序列化与反序列化

Pydantic模型提供了便捷的方法来将实例序列化为JSON字符串，以及从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到模型

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

AppMall应用商店

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()

登录后复制

注意事项：

错误处理： 在实际应用中，务必添加健壮的错误处理机制，例如文件不存在、JSON格式错误等情况，确保应用不会崩溃。
文件编码： 使用 encoding="utf-8" 确保处理各种字符集。
默认状态： 当加载失败或文件不存在时，返回一个默认的 ApplicationState 实例，保证应用始终处于可用状态。

集成到Streamlit：使用on_change回调

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

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

import streamlit as st
import os
import json
from typing import List, Optional
from 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("请尝试修改上面的选项，然后刷新页面或关闭再打开应用，查看状态是否被保留。")

登录后复制

关键点说明：

st.session_state 的使用： 我们将 ApplicationState 实例存储在 st.session_state.app_state 中。这是因为Streamlit每次运行脚本时都会重新初始化，st.session_state是跨脚本重新运行保持变量值的唯一方式。
on_change 回调： 每个需要持久化的Streamlit组件都应绑定一个on_change回调函数。
args 参数： on_change回调函数可以通过args参数接收额外的参数。这里我们传递了状态字段的key和组件的当前值（通过st.session_state.<widget_key>获取），以便update_and_save_state函数知道如何更新ApplicationState实例。
update_and_save_state 逻辑： 这个函数负责根据传入的key和new_value更新st.session_state.app_state中的对应字段，然后调用save_application_state将整个更新后的状态保存到JSON文件。

总结与最佳实践

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

回顾与最佳实践：

Pydantic定义状态： 使用Pydantic模型清晰、结构化地定义所有需要持久化的应用参数。
model_dump_json() 和 model_validate_json()： 利用Pydantic提供的内置方法进行高效的JSON序列化和反序列化。
封装保存/加载逻辑： 将状态的读写操作封装到独立的函数中，提高代码可维护性。
st.session_state 结合： 在Streamlit应用中，使用st.session_state来持有ApplicationState实例，确保在脚本重新运行时状态不会丢失。
on_change 回调： 将状态保存函数绑定到Streamlit组件的on_change事件，实现用户交互时的自动保存。
错误处理： 在文件操作和JSON解析中加入try-except块，提高应用的鲁棒性。
状态文件位置： 考虑将状态文件放置在用户可访问且权限合适的目录，例如用户的主目录或应用数据目录。对于生产环境，可能需要更复杂的存储方案（如数据库、云存储）。
性能考量： 对于非常频繁的状态更新或非常大的状态对象，频繁地读写文件可能会影响性能。在这种情况下，可以考虑引入一个延迟保存机制（例如，在用户停止操作几秒后才保存），或者只保存变化的最小子集。