在 Streamlit 中使用 SQLModel

1. 联表定义

注意：

修改 model 一定要删除 db，不然不会生效
必要的时候要重启 streamlit

模型定义


class Task_Item(SQLModel, table=True):
    __table_args__ = {"extend_existing": True}
 
    id: Optional[int] = Field(default=None, primary_key=True)
    group_id: int = Field(foreign_key="task_group.id")  # 修改外键引用
    created_at: datetime = Field(default_factory=datetime.utcnow)
    group: "Task_Group" = Relationship(back_populates="items")
 
 
class Task_Group(SQLModel, table=True):
    __table_args__ = {"extend_existing": True}
 
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    created_at: datetime = Field(default_factory=datetime.utcnow)
    items: List["Task_Item"] = Relationship(back_populates="group")

使用地方


def create_group(name):
    with Session(engine) as session:
        created_at = datetime.now()
        group = Task_Group(name=name, created_at=created_at)
        session.add(group)
        session.commit()
        session.refresh(group)
        return group.id

2. 列支持 JSON


class Task(SQLModel, table=True):
    __table_args__ = {"extend_existing": True}
    id: int = Field(default=None, primary_key=True)
    base_data: Optional[Dict[str, Any]] = Field(sa_column=Column(JSON))  # 使用 JSON 格式存储
 
# 配置json_serializer用来避免中文乱码
engine = create_engine(
    "sqlite:///myla.db", json_serializer=lambda obj: json.dumps(obj, ensure_ascii=False)
)

3. 支持多个 sqlite 数据库


from sqlmodel import SQLModel, MetaData
class SQLModel1(SQLModel):
    metadata = MetaData()
 
class SQLModel2(SQLModel):
    metadata = MetaData()
 
SQLModel1.metadata.create_all(engine1)
SQLModel2.metadata.create_all(engine2)