Pydantic CRUD

project_portfolio.py

import asyncio
from livingcharter._subcommands import create_yaml
import os

import streamlit as st
import yaml

import streamlit_pydantic as sp

from typing import List, Dict
from pydantic import BaseModel, Field
import datetime

from utils.create_prompts import create_data


class Task(BaseModel):
    title: str = Field(..., description="Title of the task")
    description: str = Field(..., description="Description of the task")
    due_date: datetime.date = Field(..., description="Due date of the task")
    status: str = Field(..., description="Status of the task")


class Project(BaseModel):
    name: str = Field(..., description="Name of the project")
    description: str = Field(..., description="Description of the project")
    start_date: datetime.date = Field(..., description="Start date of the project")
    end_date: datetime.date = Field(None, description="End date of the project")
    status: str = Field(..., description="Current status of the project")
    tasks: List[Task] = Field(..., description="List of tasks in the")

    class Config:
        allow_extra = True


from_model_tab, from_instance_tab = st.tabs(
    ["Form inputs from model", "Form inputs from instance"]
)


def to_yaml(model: BaseModel, data, file_path: str = None) -> str:
    yaml_content = yaml.dump(data, default_flow_style=False, width=1000)
    if file_path:
        with open(file_path, "w") as yaml_file:
            yaml_file.write(yaml_content)
    return yaml_content


def from_yaml(model_cls: BaseModel, file_path: str) -> BaseModel:
    with open(file_path, "r") as yaml_file:
        data = yaml.safe_load(yaml_file)
    return model_cls(**data)


data_state_key = "Project-data"

# Check if there's data in the session state
if data_state_key in st.session_state:
    data_state = st.session_state[data_state_key]
    print(
        "Found data",
    )
    instance = Project.parse_obj(data_state)
    to_yaml(Project, data_state, "Project.yaml")

if os.path.exists("Project.yaml"):
    instance = from_yaml(Project, "Project.yaml")
else:
    instance = Project

# Render the Pydantic form and update the instance

# data = sp.pydantic_form(key=f"Project", model=self.instance)

data_prompt = st.text_area("Data Prompt", "")

# Add a button to trigger the conversion
if st.button("Convert to Data"):
    st.write("Converting...")

    # Run the conversion asynchronously
    async def async_conversion():
        project_data = await create_data(data_prompt, Project)
        st.json(project_data)

    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(async_conversion())
    st.write("Conversion complete!")


yaml_prompt = st.text_area("Yaml Prompt", "")

# Add a button to trigger the conversion
if st.button("Convert to Yaml"):
    st.write("Converting...")

    # Run the conversion asynchronously
    async def async_conversion():
        yaml_data = await create_yaml.render(yaml_prompt)
        st.code(yaml_data, language="yaml")

    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(async_conversion())
    st.write("Conversion complete!")


with from_model_tab:
    data = sp.pydantic_input(key="Project", model=instance)
    with st.expander("Current Input State", expanded=False):
        st.json(data)
        to_yaml(Project, data, "Project.yaml")

st.markdown("---")

if data:
    instance = data
    print(type(instance))
    # st.json(data.json())

with st.expander("Session State", expanded=False):
    st.write(st.session_state)