-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathmain.py
More file actions
162 lines (132 loc) · 5.11 KB
/
main.py
File metadata and controls
162 lines (132 loc) · 5.11 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
import os
from flask import Flask, render_template, request, redirect, url_for, flash, jsonify
import sqlite3
from datetime import datetime
app = Flask(__name__)
app.secret_key = os.environ.get("SECRET_KEY", "dev-secret-key-change-me")
DATABASE = os.environ.get("DATABASE_PATH", "database.db")
def get_db():
conn = sqlite3.connect(DATABASE)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
return conn
def init_db():
with get_db() as db:
db.execute("""
CREATE TABLE IF NOT EXISTS items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
description TEXT DEFAULT '',
status TEXT DEFAULT 'active' CHECK(status IN ('active','completed','archived')),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_items_status ON items(status)
""")
init_db()
# ── API Routes ──────────────────────────────────────────────────────────────
@app.route("/")
def index():
filter_status = request.args.get("status", "all")
search = request.args.get("q", "").strip()
db = get_db()
query = "SELECT * FROM items"
params = []
conditions = []
if filter_status != "all":
conditions.append("status = ?")
params.append(filter_status)
if search:
conditions.append("(title LIKE ? OR description LIKE ?)")
params.extend([f"%{search}%", f"%{search}%"])
if conditions:
query += " WHERE " + " AND ".join(conditions)
query += " ORDER BY created_at DESC"
items = db.execute(query, params).fetchall()
db.close()
return render_template("index.html", items=items, filter_status=filter_status, search=search)
@app.route("/create", methods=["POST"])
def create():
title = request.form.get("title", "").strip()
description = request.form.get("description", "").strip()
if not title:
flash("Title is required.", "error")
return redirect(url_for("index"))
db = get_db()
db.execute("INSERT INTO items (title, description) VALUES (?, ?)", (title, description))
db.commit()
db.close()
flash("Item created successfully.", "success")
return redirect(url_for("index"))
@app.route("/update/<int:item_id>", methods=["POST"])
def update(item_id):
title = request.form.get("title", "").strip()
description = request.form.get("description", "").strip()
status = request.form.get("status", "active")
if not title:
flash("Title is required.", "error")
return redirect(url_for("index"))
db = get_db()
db.execute(
"UPDATE items SET title=?, description=?, status=?, updated_at=CURRENT_TIMESTAMP WHERE id=?",
(title, description, status, item_id),
)
db.commit()
db.close()
flash("Item updated.", "success")
return redirect(url_for("index"))
@app.route("/delete/<int:item_id>", methods=["POST"])
def delete(item_id):
db = get_db()
db.execute("DELETE FROM items WHERE id=?", (item_id,))
db.commit()
db.close()
flash("Item deleted.", "success")
return redirect(url_for("index"))
@app.route("/toggle/<int:item_id>", methods=["POST"])
def toggle(item_id):
db = get_db()
item = db.execute("SELECT status FROM items WHERE id=?", (item_id,)).fetchone()
if item:
new_status = "completed" if item["status"] == "active" else "active"
db.execute(
"UPDATE items SET status=?, updated_at=CURRENT_TIMESTAMP WHERE id=?",
(new_status, item_id),
)
db.commit()
db.close()
return redirect(url_for("index"))
# ── JSON API (bonus for learners) ──────────────────────────────────────────
@app.route("/api/items", methods=["GET"])
def api_list():
db = get_db()
items = db.execute("SELECT * FROM items ORDER BY created_at DESC").fetchall()
db.close()
return jsonify([dict(i) for i in items])
@app.route("/api/items", methods=["POST"])
def api_create():
data = request.get_json(force=True)
title = data.get("title", "").strip()
if not title:
return jsonify({"error": "Title is required"}), 400
db = get_db()
cur = db.execute(
"INSERT INTO items (title, description) VALUES (?, ?)",
(title, data.get("description", "")),
)
db.commit()
item = db.execute("SELECT * FROM items WHERE id=?", (cur.lastrowid,)).fetchone()
db.close()
return jsonify(dict(item)), 201
@app.route("/api/items/<int:item_id>", methods=["DELETE"])
def api_delete(item_id):
db = get_db()
db.execute("DELETE FROM items WHERE id=?", (item_id,))
db.commit()
db.close()
return jsonify({"deleted": item_id})
if __name__ == "__main__":
port = int(os.environ.get("PORT", 8080))
app.run(host="0.0.0.0", port=port, debug=os.environ.get("DEBUG", "false").lower() == "true")