Upload app.py with huggingface_hub

app.py

@@ -1,19 +1,23 @@
 # -*- coding: utf-8 -*-
 import os
 import sqlite3
-import gradio as gr
+from flask import Flask, request, jsonify, g
 
-DB_PATH = os.path.join(os.path.dirname(__file__), "tasks.db")
+app = Flask(__name__)
 
-def get_connection():
-    conn = sqlite3.connect(DB_PATH)
-    conn.row_factory = sqlite3.Row
-    return conn
+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 init_db():
-    conn = get_connection()
-    cur = conn.cursor()
-    cur.execute(
+    conn = sqlite3.connect(DATABASE)
+    cursor = conn.cursor()
+    cursor.execute(
         """
         CREATE TABLE IF NOT EXISTS tasks (
             id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -25,104 +29,68 @@ def init_db():
     conn.commit()
     conn.close()
 
-def fetch_all():
-    conn = get_connection()
-    cur = conn.cursor()
-    cur.execute("SELECT id, title, completed FROM tasks")
-    rows = cur.fetchall()
-    conn.close()
-    return [[r["id"], r["title"], bool(r["completed"])] for r in rows]
-
-def get_task_options():
-    """Return list of task ids as strings for dropdowns."""
-    tasks = fetch_all()
-    return [str(row[0]) for row in tasks]
-
-def refresh_tasks():
-    """Return only the data for the Dataframe component."""
-    return fetch_all()
-
-def add_task(title):
+@app.teardown_appcontext
+def close_connection(exception):
+    db = getattr(g, "_database", None)
+    if db is not None:
+        db.close()
+
+@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)
+
+@app.route("/tasks", methods=["POST"])
+def create_task():
+    data = request.get_json()
+    title = data.get("title")
     if not title:
-        return refresh_tasks()
-    conn = get_connection()
-    cur = conn.cursor()
-    cur.execute("INSERT INTO tasks (title, completed) VALUES (?, ?)", (title, 0))
-    conn.commit()
-    conn.close()
-    return refresh_tasks()
-
-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)),
+        return jsonify({"error": "title is required"}), 400
+    db = get_db()
+    cursor = db.execute(
+        "INSERT INTO tasks (title, completed) VALUES (?, ?)",
+        (title, 0),
     )
-    conn.commit()
-    conn.close()
-    return refresh_tasks()
-
-def delete_task(task_id):
-    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()
-
-# Initialise DB 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"],
-            value=[],
-            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(),
+    db.commit()
+    task_id = cursor.lastrowid
+    return jsonify({"id": task_id, "title": title, "completed": False}), 201
+
+@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),
         )
-        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)
+    db.commit()
+    return jsonify({"id": task_id, "title": title, "completed": completed})
 
-    add_btn.click(fn=add_task, inputs=add_title, outputs=task_table).then(
-        fn=get_task_options, inputs=None, outputs=[upd_id, del_id]
-    )
-
-    upd_btn.click(fn=update_task, inputs=[upd_id, upd_title, upd_completed], outputs=task_table).then(
-        fn=get_task_options, inputs=None, outputs=[upd_id, del_id]
-    )
-
-    del_btn.click(fn=delete_task, inputs=del_id, outputs=task_table).then(
-        fn=get_task_options, inputs=None, outputs=[upd_id, del_id]
-    )
-
-demo.launch()
+@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
+
+@app.route("/", methods=["GET"])
+def index():
+    return "Task API is running."
+
+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)