timmy / SQLite 與 Streamlit st.data_editor 整合

import sqlite3 import pandas as pd import streamlit as st from datetime import datetime, date def initialize_database(conn): cursor = conn.cursor() # 建立 Customers 資料表 cursor.execute(''' CREATE TABLE IF NOT EXISTS Customers ( CustomerID INTEGER PRIMARY KEY, CustomerName TEXT NOT NULL, ContactName TEXT, Country TEXT ) ''') # 建立 Orders 資料表 cursor.execute(''' CREATE TABLE IF NOT EXISTS Orders ( OrderID INTEGER PRIMARY KEY, CustomerID INTEGER, OrderDate TEXT, FOREIGN KEY (CustomerID) REFERENCES Customers(CustomerID) ) ''') # 插入範例資料到 Customers（如果表格為空） cursor.execute('SELECT COUNT(*) FROM Customers') if cursor.fetchone()[0] == 0: customers = [ (1, 'Alfreds Futterkiste', 'Maria Anders', 'Germany'), (2, 'Ana Trujillo Emparedados y helados', 'Ana Trujillo', 'Mexico'), (3, 'Antonio Moreno Taquería', 'Antonio Moreno', 'Mexico'), (4, 'Around the Horn', 'Thomas Hardy', 'UK') ] cursor.executemany('INSERT INTO Customers VALUES (?, ?, ?, ?)', customers) # 插入範例資料到 Orders（如果表格為空） cursor.execute('SELECT COUNT(*) FROM Orders') if cursor.fetchone()[0] == 0: orders = [ (101, 1, '2023-11-01'), (102, 2, '2023-11-03'), (103, 1, '2023-11-07'), (104, 3, '2023-11-10'), (105, 4, '2023-11-12') ] cursor.executemany('INSERT INTO Orders VALUES (?, ?, ?)', orders) conn.commit() def get_joined_data(conn): query = ''' SELECT Customers.CustomerID, Customers.CustomerName, Orders.OrderID, Orders.OrderDate FROM Customers INNER JOIN Orders ON Customers.CustomerID = Orders.CustomerID ORDER BY Customers.CustomerID ''' df = pd.read_sql_query(query, conn) # 將 OrderDate 轉換為 datetime 類型 df['OrderDate'] = pd.to_datetime(df['OrderDate']).dt.date return df def main(): st.title("SQLite JOIN 範例與 Streamlit st.data_editor") # 連接到 SQLite 資料庫（這裡使用本地檔案資料庫） conn = sqlite3.connect('example.db', check_same_thread=False) initialize_database(conn) # 獲取 JOIN 查詢的資料 df = get_joined_data(conn) st.header("JOIN 查詢結果") edited_df = st.data_editor( df, num_rows="dynamic", key="data_editor", use_container_width=True, column_config={ "CustomerID": st.column_config.NumberColumn("Customer ID", disabled=True), "CustomerName": st.column_config.TextColumn("客戶名稱", disabled=True), # 禁用編輯 "OrderID": st.column_config.NumberColumn("訂單編號", disabled=True), "OrderDate": st.column_config.DateColumn("訂單日期") } ) if st.button("更新資料庫"): # 將編輯後的資料寫回資料庫 cursor = conn.cursor() updates = [] errors = [] for index, row in edited_df.iterrows(): # 確保 OrderDate 是 datetime.date 類型，轉換為字串格式 if isinstance(row['OrderDate'], datetime): order_date_str = row['OrderDate'].strftime('%Y-%m-%d') elif isinstance(row['OrderDate'], pd.Timestamp): order_date_str = row['OrderDate'].date().strftime('%Y-%m-%d') elif isinstance(row['OrderDate'], date): order_date_str = row['OrderDate'].strftime('%Y-%m-%d') elif isinstance(row['OrderDate'], str): # 嘗試解析字串為日期 try: order_date = datetime.strptime(row['OrderDate'], '%Y-%m-%d').date() order_date_str = order_date.strftime('%Y-%m-%d') except ValueError: errors.append(f"無效的日期格式: {row['OrderDate']} (訂單編號: {row['OrderID']})") continue else: errors.append(f"無法解析的日期格式: {row['OrderDate']} (訂單編號: {row['OrderID']})") continue updates.append((order_date_str, row['OrderID'])) if updates: try: cursor.executemany(''' UPDATE Orders SET OrderDate = ? WHERE OrderID = ? ''', updates) conn.commit() st.success("資料庫已更新！") except Exception as e: st.error(f"資料庫更新失敗: {e}") if errors: for error in errors: st.error(error) # 不需要手動重新執行腳本，Streamlit 會自動重新渲染並顯示更新後的資料 # 關閉資料庫連接 conn.close() if __name__ == "__main__": main()