Upload app.py with huggingface_hub

app.py

@@ -1,23 +1,19 @@
 # -*- coding: utf-8 -*-
 import os
 import sqlite3
-from flask import Flask, request, jsonify, g
+import gradio as gr
 
-app = Flask(__name__)
+DB_PATH = os.path.join(os.path.dirname(__file__), "tasks.db")
 
-DATABASE = os.path.join(os.path.dirname(__file__), "tasks.db")
-
-def get_db():
-    db = getattr(g, "_database", None)
-    if db is None:
-        db = g._database = sqlite3.connect(DATABASE)
-        db.row_factory = sqlite3.Row
-    return db
+def get_connection():
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    return conn
 
 def init_db():
-    conn = sqlite3.connect(DATABASE)
-    cursor = conn.cursor()
-    cursor.execute(
+    conn = get_connection()
+    cur = conn.cursor()
+    cur.execute(
         """
         CREATE TABLE IF NOT EXISTS tasks (
             id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -29,64 +25,89 @@ def init_db():
     conn.commit()
     conn.close()
 
-@app.teardown_appcontext
-def close_connection(exception):
-    db = getattr(g, "_database", None)
-    if db is not None:
-        db.close()
+def fetch_all():
+    conn = get_connection()
+    cur = conn.cursor()
+    cur.execute("SELECT id, title, completed FROM tasks")
+    rows = cur.fetchall()
+    conn.close()
+    data = [[r["id"], r["title"], bool(r["completed"])] for r in rows]
+    return data
+
+def get_task_options():
+    data = fetch_all()
+    return [str(row[0]) for row in data]
 
-@app.route("/tasks", methods=["GET"])
-def list_tasks():
-    db = get_db()
-    cursor = db.execute("SELECT id, title, completed FROM tasks")
-    tasks = [
-        {"id": row["id"], "title": row["title"], "completed": bool(row["completed"])}
-        for row in cursor.fetchall()
-    ]
-    return jsonify(tasks)
+def refresh_tasks():
+    data = fetch_all()
+    return data, ["id", "title", "completed"]
 
-@app.route("/tasks", methods=["POST"])
-def create_task():
-    data = request.get_json()
-    title = data.get("title")
+def add_task(title):
     if not title:
-        return jsonify({"error": "title is required"}), 400
-    db = get_db()
-    cursor = db.execute(
-        "INSERT INTO tasks (title, completed) VALUES (?, ?)",
-        (title, 0),
-    )
-    db.commit()
-    task_id = cursor.lastrowid
-    return jsonify({"id": task_id, "title": title, "completed": False}), 201
+        return fetch_all(), ["id", "title", "completed"]
+    conn = get_connection()
+    cur = conn.cursor()
+    cur.execute("INSERT INTO tasks (title, completed) VALUES (?, ?)", (title, 0))
+    conn.commit()
+    conn.close()
+    return refresh_tasks()
 
-@app.route("/tasks/<int:task_id>", methods=["PUT"])
-def update_task(task_id):
-    data = request.get_json()
-    title = data.get("title")
-    completed = data.get("completed")
-    if title is None and completed is None:
-        return jsonify({"error": "nothing to update"}), 400
-    db = get_db()
-    if title is not None:
-        db.execute("UPDATE tasks SET title = ? WHERE id = ?", (title, task_id))
-    if completed is not None:
-        db.execute(
-            "UPDATE tasks SET completed = ? WHERE id = ?",
-            (1 if completed else 0, task_id),
-        )
-    db.commit()
-    return jsonify({"id": task_id, "title": title, "completed": completed})
+def update_task(task_id, new_title, completed):
+    if not task_id:
+        return refresh_tasks()
+    conn = get_connection()
+    cur = conn.cursor()
+    if new_title:
+        cur.execute("UPDATE tasks SET title = ? WHERE id = ?", (new_title, int(task_id)))
+    cur.execute("UPDATE tasks SET completed = ? WHERE id = ?", (1 if completed else 0, int(task_id)))
+    conn.commit()
+    conn.close()
+    return refresh_tasks()
 
-@app.route("/tasks/<int:task_id>", methods=["DELETE"])
 def delete_task(task_id):
-    db = get_db()
-    db.execute("DELETE FROM tasks WHERE id = ?", (task_id,))
-    db.commit()
-    return "", 204
+    if not task_id:
+        return refresh_tasks()
+    conn = get_connection()
+    cur = conn.cursor()
+    cur.execute("DELETE FROM tasks WHERE id = ?", (int(task_id),))
+    conn.commit()
+    conn.close()
+    return refresh_tasks()
+
+# Initialize database on first run
+if not os.path.isfile(DB_PATH):
+    init_db()
+
+with gr.Blocks() as demo:
+    gr.Markdown("# Docker Task Manager")
+    with gr.Row():
+        task_table = gr.Dataframe(
+            label="Tasks",
+            headers=["id", "title", "completed"],
+            datatype=["number", "str", "bool"],
+            interactive=False,
+        )
+        refresh_btn = gr.Button("Refresh")
+    with gr.Row():
+        add_title = gr.Textbox(label="New Task Title", placeholder="Enter task title")
+        add_btn = gr.Button("Add Task")
+    with gr.Row():
+        upd_id = gr.Dropdown(label="Task ID to Update", choices=get_task_options())
+        upd_title = gr.Textbox(label="New Title (optional)", placeholder="Leave empty to keep current")
+        upd_completed = gr.Checkbox(label="Completed")
+        upd_btn = gr.Button("Update Task")
+    with gr.Row():
+        del_id = gr.Dropdown(label="Task ID to Delete", choices=get_task_options())
+        del_btn = gr.Button("Delete Task")
+
+    # Bind actions
+    refresh_btn.click(fn=refresh_tasks, inputs=None, outputs=[task_table])
+    add_btn.click(fn=add_task, inputs=add_title, outputs=[task_table])
+    upd_btn.click(fn=update_task, inputs=[upd_id, upd_title, upd_completed], outputs=[task_table])
+    del_btn.click(fn=delete_task, inputs=del_id, outputs=[task_table])
+
+    # Update dropdown options after any change
+    for btn in [add_btn, upd_btn, del_btn]:
+        btn.click(fn=get_task_options, inputs=None, outputs=[upd_id, del_id])
 
-if __name__ == "__main__":
-    # Ensure the database exists before the first request
-    if not os.path.isfile(DATABASE):
-        init_db()
-    app.run(host="0.0.0.0", port=7860)
+demo.launch()