Streamlit表格列类型一致性问题：Optional字段致追加行失败

问题背景 #

需要将Pydantic BaseModel子类的序列化数据行追加到Streamlit表格中，但部分字段为Optional类型，存在缺失情况。尝试通过初始化pandas.DataFrame时指定列类型和设置column_config两种方式预定义列类型均无效，运行时出现类型不匹配报错：

Unsupported operation. The data passed into add_rows() must have the same data signature as the original data.

In this case, add_rows() received ["unicode","unicode","unicode","unicode","unicode","unicode","unicode","unicode","unicode","unicode","datetime","unicode"]
but was expecting ["empty","empty","empty","empty","empty","empty","empty","empty","empty","empty","datetime","empty"].

或：

elementType 'alert' is not a valid arrowAddRows target!

错误原因 #

1. 初始化表格时使用np.empty(0)生成空数组，Streamlit会将这类空列识别为empty类型，与后续追加的实际数据类型（如unicode/datetime）不匹配。
2. 重复执行st.session_state['df_reports'] = st_from_base_model(MyModel)会覆盖之前的表格组件对象，导致add_rows调用目标失效。
3. Optional字段的类型转换逻辑存在疏漏，未确保初始化和追加数据的列类型完全对齐。

解决方案 #

核心思路 #

• 初始化表格时使用带正确列类型的结构完整空DataFrame，让Streamlit识别到正确的列类型。
• 统一BaseModel到DataFrame的类型转换逻辑，确保初始化和追加数据的列类型完全一致。
• 正确管理Streamlit表格组件的引用，避免重复初始化覆盖对象。

修改后的代码实现 #

import pandas as pd
from pydantic import BaseModel
import streamlit as st
from functools import partial
from typing import Optional, List
import datetime

class MyModel(BaseModel):
    first: str 
    second: Optional[str]
    date: datetime.datetime = datetime.datetime.today()

# 定义Pydantic类型到Streamlit列配置的映射
TYPE_TO_STREAMLIT = {
    "str": st.column_config.TextColumn,
    "float": st.column_config.NumberColumn,
    "int": partial(st.column_config.NumberColumn, format='%u'),
    "datetime64[ns]": partial(st.column_config.DatetimeColumn, format='YYYY-MM-DD HH:mm:ss'),
}

def get_pandas_type_from_field(field) -> str:
    """将Pydantic字段类型转换为Pandas可识别的类型字符串"""
    field_type = field.annotation
    # 处理Optional类型
    if hasattr(field_type, "__origin__") and field_type.__origin__ is Optional:
        inner_type = field_type.__args__[0]
        if inner_type is str:
            return "str"
        elif inner_type is datetime.datetime:
            return "datetime64[ns]"
        else:
            return str(inner_type.__name__)
    # 处理非Optional的基础类型
    elif field_type is str:
        return "str"
    elif field_type is datetime.datetime:
        return "datetime64[ns]"
    else:
        return str(field_type.__name__)

def create_empty_df_from_model(model: BaseModel) -> pd.DataFrame:
    """基于BaseModel创建带正确列类型的空DataFrame"""
    col_types = {
        field_name: get_pandas_type_from_field(field)
        for field_name, field in model.model_fields.items()
    }
    # 创建一行空数据（用NaN填充）来固定列类型，然后删除这行得到空DataFrame
    empty_row = {col: None for col in col_types.keys()}
    df = pd.DataFrame([empty_row]).astype(col_types)
    return df.iloc[0:0]

def get_streamlit_column_config(model: BaseModel) -> dict:
    """生成Streamlit表格的列配置"""
    return {
        field_name: TYPE_TO_STREAMLIT[get_pandas_type_from_field(field)](field_name.replace('_', ' '))
        for field_name, field in model.model_fields.items()
    }

def convert_models_to_df(models: List[BaseModel]) -> pd.DataFrame:
    """将BaseModel列表转换为带正确类型的DataFrame"""
    if not models:
        return create_empty_df_from_model(MyModel)
    # 先dump数据，再统一转换类型
    data = [md.model_dump() for md in models]
    df = pd.DataFrame(data)
    col_types = {
        field_name: get_pandas_type_from_field(field)
        for field_name, field in MyModel.model_fields.items()
    }
    return df.astype(col_types)

# 主逻辑
if 'df_reports' not in st.session_state:
    # 初始化空DataFrame到session_state
    st.session_state['df_reports'] = create_empty_df_from_model(MyModel)

# 渲染表格并获取组件引用
table = st.dataframe(
    st.session_state['df_reports'],
    column_config=get_streamlit_column_config(MyModel),
    use_container_width=True
)

# 示例数据生成
examples = [
    MyModel(**{"first":x, "second":chr(ord(x)+5) if ord(x)%2==0 else None}) 
    for x in "hello world!"
]

# 追加数据按钮（避免每次刷新都重复追加）
if st.button("追加示例数据"):
    new_df = convert_models_to_df(examples)
    table.add_rows(new_df)
    # 更新session_state中的DataFrame（可选，用于后续其他操作）
    st.session_state['df_reports'] = pd.concat([st.session_state['df_reports'], new_df], ignore_index=True)

关键修改点说明 #

1. 初始化DataFrame方式：通过创建一行空数据再删除，确保Pandas和Streamlit能识别正确的列类型，替代原有的np.empty(0)方式。
2. 类型转换逻辑：优化get_pandas_type_from_field函数，更准确地处理Optional类型和datetime类型的映射，确保初始化和追加数据的列类型完全一致。
3. 组件引用管理：只初始化一次session_state中的DataFrame，渲染表格时获取组件引用table，后续通过该引用调用add_rows，避免重复初始化导致的对象失效。
4. 避免重复追加：使用按钮触发追加操作，防止Streamlit每次刷新都重复添加数据。

内容的提问来源于stack exchange，提问作者Dima Lituiev