| |
| import os |
| import sqlite3 |
| from flask import Flask, request, jsonify, g |
|
|
| app = Flask(__name__) |
|
|
| 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 = sqlite3.connect(DATABASE) |
| cursor = conn.cursor() |
| cursor.execute( |
| """ |
| CREATE TABLE IF NOT EXISTS tasks ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| title TEXT NOT NULL, |
| completed INTEGER NOT NULL DEFAULT 0 |
| ) |
| """ |
| ) |
| conn.commit() |
| conn.close() |
|
|
| @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 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 |
|
|
| @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}) |
|
|
| @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__": |
| |
| if not os.path.isfile(DATABASE): |
| init_db() |
| |
| import subprocess |
| app.run(host="0.0.0.0", port=7860) |