mysite/routers/database.py

import duckdb
import pandas as pd
from fastapi import FastAPI
import gradio as gr

con = duckdb.connect(database="./workspace/mydatabase.duckdb")
con.execute("CREATE TABLE IF NOT EXISTS items (id INTEGER, name VARCHAR);")

# Extract the 'content' field from all elements in the result
def insert(full_response,message):
    age = 28
    # データベースファイルのパス
    db_path = "./workspace/sample.duckdb"

    # DuckDBに接続（データベースファイルが存在しない場合は新規作成）
    con = duckdb.connect(database=db_path)
    con.execute(
        """
    CREATE SEQUENCE IF NOT EXISTS sample_id_seq START 1;
    CREATE TABLE IF NOT EXISTS samples (
        id INTEGER DEFAULT nextval('sample_id_seq'),
        name VARCHAR,
        age INTEGER,
        PRIMARY KEY(id)
    );
    """
    )
    cur = con.cursor()
    con.execute("INSERT INTO samples (name, age) VALUES (?, ?)", (full_response, age))
    con.execute("INSERT INTO samples (name, age) VALUES (?, ?)", (message, age))
    # データをCSVファイルにエクスポート
    con.execute("COPY samples TO 'sample.csv' (FORMAT CSV, HEADER)")
    # データをコミット
    con.commit()
    # データを選択
    cur = con.execute("SELECT * FROM samples")
    # 結果をフェッチ
    res = cur.fetchall()
    rows = ""
    # 結果を表示
    # 結果を文字列に整形
    rows = "\n".join([f"name: {row[0]}, age: {row[1]}" for row in res])
    # コネクションを閉じる
    con.close()
    # print(cur.fetchall())
    insert(full_response,message)

def setup_database_routes(app: FastAPI):
    def create_item(name):
        con.execute("INSERT INTO items (name) VALUES (?);", (name,))
        con.commit()
        return "Item created successfully!"

    def read_items():
        cursor = con.cursor()
        cursor.execute("SELECT * FROM items;")
        items = cursor.fetchall()
        df = pd.DataFrame(items, columns=["ID", "Name"])
        return df

    def update_item(id, name):
        con.execute("UPDATE items SET name = ? WHERE id = ?;", (name, id))
        con.commit()
        return "Item updated successfully!"

    def delete_item(id):
        con.execute("DELETE FROM items WHERE id = ?;", (id,))
        con.commit()
        return "Item deleted successfully!"

    with gr.Blocks() as appdb:
        gr.Markdown("CRUD Application")
        with gr.Row():
            with gr.Column():
                create_name = gr.Textbox(label="Create Item")
                create_btn = gr.Button("Create")
            with gr.Column():
                read_btn = gr.Button("Read Items")
        with gr.Row():
            with gr.Column():
                update_id = gr.Textbox(label="Update Item ID")
                update_name = gr.Textbox(label="Update Item Name")
                update_btn = gr.Button("Update")
            with gr.Column():
                delete_id = gr.Textbox(label="Delete Item ID")
                delete_btn = gr.Button("Delete")
        output_text = gr.Textbox(label="Output")
        output_table = gr.DataFrame(label="Items")

        create_btn.click(fn=create_item, inputs=create_name, outputs=output_text)
        read_btn.click(fn=read_items, outputs=output_table)
        update_btn.click(fn=update_item, inputs=[update_id, update_name], outputs=output_text)
        delete_btn.click(fn=delete_item, inputs=delete_id, outputs=output_text)

    app.mount("/db", appdb, name="database_app")