first commit
This commit is contained in:
331
server.py
Normal file
331
server.py
Normal file
@@ -0,0 +1,331 @@
|
||||
import os
|
||||
import secrets
|
||||
import sqlite3
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from dotenv import load_dotenv
|
||||
from flask import (
|
||||
Flask,
|
||||
abort,
|
||||
g,
|
||||
jsonify,
|
||||
redirect,
|
||||
render_template,
|
||||
request,
|
||||
session,
|
||||
url_for,
|
||||
)
|
||||
|
||||
load_dotenv()
|
||||
|
||||
|
||||
def create_app() -> Flask:
|
||||
app = Flask(__name__, static_folder="static", template_folder="templates")
|
||||
|
||||
secret_key = os.environ.get("SECRET_KEY")
|
||||
if not secret_key:
|
||||
# Safe default for local dev; in production set SECRET_KEY in .env
|
||||
secret_key = secrets.token_urlsafe(32)
|
||||
app.secret_key = secret_key
|
||||
|
||||
app.config["DB_PATH"] = os.environ.get("DB_PATH", "./data/app.db")
|
||||
app.config["AUTH_USERNAME"] = os.environ.get("AUTH_USERNAME", "")
|
||||
app.config["AUTH_PASSWORD"] = os.environ.get("AUTH_PASSWORD", "")
|
||||
|
||||
# Ensure DB directory exists.
|
||||
# - If DB_PATH is a bare filename (no directory), dirname() is "".
|
||||
# - If DB_PATH is absolute (e.g. /data/app.db), dirname() is "/data".
|
||||
db_dir = os.path.dirname(app.config["DB_PATH"]) or "."
|
||||
os.makedirs(db_dir, exist_ok=True)
|
||||
|
||||
@app.before_request
|
||||
def _open_db():
|
||||
g.db = sqlite3.connect(app.config["DB_PATH"])
|
||||
g.db.row_factory = sqlite3.Row
|
||||
|
||||
@app.after_request
|
||||
def _close_db(resp):
|
||||
db = getattr(g, "db", None)
|
||||
if db is not None:
|
||||
db.close()
|
||||
return resp
|
||||
|
||||
def init_db() -> None:
|
||||
db = sqlite3.connect(app.config["DB_PATH"])
|
||||
try:
|
||||
db.executescript(
|
||||
"""
|
||||
PRAGMA journal_mode=WAL;
|
||||
PRAGMA foreign_keys=ON;
|
||||
|
||||
CREATE TABLE IF NOT EXISTS todo_items (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
text TEXT NOT NULL,
|
||||
done INTEGER NOT NULL DEFAULT 0,
|
||||
created_at TEXT NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS expenses (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
description TEXT NOT NULL,
|
||||
jpy REAL NOT NULL,
|
||||
eur REAL NOT NULL,
|
||||
created_at TEXT NOT NULL
|
||||
);
|
||||
"""
|
||||
)
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
init_db()
|
||||
|
||||
def is_authed() -> bool:
|
||||
return bool(session.get("authed"))
|
||||
|
||||
def require_auth() -> None:
|
||||
if not is_authed():
|
||||
abort(401)
|
||||
|
||||
@app.get("/")
|
||||
def index():
|
||||
if not is_authed():
|
||||
return redirect(url_for("login"))
|
||||
return render_template("index.html")
|
||||
|
||||
@app.get("/login")
|
||||
def login():
|
||||
if is_authed():
|
||||
return redirect(url_for("index"))
|
||||
return render_template("login.html")
|
||||
|
||||
@app.post("/login")
|
||||
def login_post():
|
||||
username = (request.form.get("username") or "").strip()
|
||||
password = request.form.get("password") or ""
|
||||
|
||||
if not app.config["AUTH_USERNAME"] or not app.config["AUTH_PASSWORD"]:
|
||||
# Misconfiguration: refuse login if credentials are not set.
|
||||
abort(500)
|
||||
|
||||
if username == app.config["AUTH_USERNAME"] and password == app.config["AUTH_PASSWORD"]:
|
||||
session.clear()
|
||||
session["authed"] = True
|
||||
return redirect(url_for("index"))
|
||||
|
||||
return render_template("login.html", error="Credenziali non valide")
|
||||
|
||||
@app.post("/logout")
|
||||
def logout():
|
||||
session.clear()
|
||||
return redirect(url_for("login"))
|
||||
|
||||
# -----------------
|
||||
# API: TODO
|
||||
# -----------------
|
||||
|
||||
@app.get("/api/todos")
|
||||
def api_todos_list():
|
||||
require_auth()
|
||||
rows = g.db.execute(
|
||||
"SELECT id, text, done, created_at FROM todo_items ORDER BY id DESC"
|
||||
).fetchall()
|
||||
return jsonify(
|
||||
[
|
||||
{
|
||||
"id": r["id"],
|
||||
"text": r["text"],
|
||||
"done": bool(r["done"]),
|
||||
"created_at": r["created_at"],
|
||||
}
|
||||
for r in rows
|
||||
]
|
||||
)
|
||||
|
||||
@app.post("/api/todos")
|
||||
def api_todos_create():
|
||||
require_auth()
|
||||
data = request.get_json(silent=True) or {}
|
||||
text = (data.get("text") or "").strip()
|
||||
if not text:
|
||||
return jsonify({"error": "text is required"}), 400
|
||||
|
||||
now = datetime.now(timezone.utc).isoformat()
|
||||
cur = g.db.execute(
|
||||
"INSERT INTO todo_items(text, done, created_at) VALUES(?, 0, ?)",
|
||||
(text, now),
|
||||
)
|
||||
g.db.commit()
|
||||
return jsonify({"id": cur.lastrowid, "text": text, "done": False, "created_at": now}), 201
|
||||
|
||||
@app.patch("/api/todos/<int:todo_id>")
|
||||
def api_todos_update(todo_id: int):
|
||||
require_auth()
|
||||
data = request.get_json(silent=True) or {}
|
||||
|
||||
fields = []
|
||||
params = []
|
||||
|
||||
if "text" in data:
|
||||
text = (data.get("text") or "").strip()
|
||||
if not text:
|
||||
return jsonify({"error": "text cannot be empty"}), 400
|
||||
fields.append("text = ?")
|
||||
params.append(text)
|
||||
|
||||
if "done" in data:
|
||||
done = 1 if bool(data.get("done")) else 0
|
||||
fields.append("done = ?")
|
||||
params.append(done)
|
||||
|
||||
if not fields:
|
||||
return jsonify({"error": "no fields to update"}), 400
|
||||
|
||||
params.append(todo_id)
|
||||
cur = g.db.execute(
|
||||
f"UPDATE todo_items SET {', '.join(fields)} WHERE id = ?",
|
||||
tuple(params),
|
||||
)
|
||||
g.db.commit()
|
||||
if cur.rowcount == 0:
|
||||
return jsonify({"error": "not found"}), 404
|
||||
return jsonify({"ok": True})
|
||||
|
||||
@app.delete("/api/todos/<int:todo_id>")
|
||||
def api_todos_delete(todo_id: int):
|
||||
require_auth()
|
||||
cur = g.db.execute("DELETE FROM todo_items WHERE id = ?", (todo_id,))
|
||||
g.db.commit()
|
||||
if cur.rowcount == 0:
|
||||
return jsonify({"error": "not found"}), 404
|
||||
return jsonify({"ok": True})
|
||||
|
||||
# -----------------
|
||||
# API: EXPENSES
|
||||
# -----------------
|
||||
|
||||
@app.get("/api/expenses")
|
||||
def api_expenses_list():
|
||||
require_auth()
|
||||
rows = g.db.execute(
|
||||
"SELECT id, description, jpy, eur, created_at FROM expenses ORDER BY id DESC"
|
||||
).fetchall()
|
||||
return jsonify(
|
||||
[
|
||||
{
|
||||
"id": r["id"],
|
||||
"description": r["description"],
|
||||
"jpy": r["jpy"],
|
||||
"eur": r["eur"],
|
||||
"created_at": r["created_at"],
|
||||
}
|
||||
for r in rows
|
||||
]
|
||||
)
|
||||
|
||||
@app.post("/api/expenses")
|
||||
def api_expenses_create():
|
||||
require_auth()
|
||||
data = request.get_json(silent=True) or {}
|
||||
description = (data.get("description") or "").strip()
|
||||
jpy = data.get("jpy")
|
||||
eur = data.get("eur")
|
||||
|
||||
if not description:
|
||||
return jsonify({"error": "description is required"}), 400
|
||||
|
||||
try:
|
||||
jpy_f = float(jpy)
|
||||
eur_f = float(eur)
|
||||
except (TypeError, ValueError):
|
||||
return jsonify({"error": "jpy and eur must be numbers"}), 400
|
||||
|
||||
now = datetime.now(timezone.utc).isoformat()
|
||||
cur = g.db.execute(
|
||||
"INSERT INTO expenses(description, jpy, eur, created_at) VALUES(?, ?, ?, ?)",
|
||||
(description, jpy_f, eur_f, now),
|
||||
)
|
||||
g.db.commit()
|
||||
return (
|
||||
jsonify(
|
||||
{
|
||||
"id": cur.lastrowid,
|
||||
"description": description,
|
||||
"jpy": jpy_f,
|
||||
"eur": eur_f,
|
||||
"created_at": now,
|
||||
}
|
||||
),
|
||||
201,
|
||||
)
|
||||
|
||||
@app.patch("/api/expenses/<int:expense_id>")
|
||||
def api_expenses_update(expense_id: int):
|
||||
require_auth()
|
||||
data = request.get_json(silent=True) or {}
|
||||
|
||||
fields = []
|
||||
params = []
|
||||
|
||||
if "description" in data:
|
||||
description = (data.get("description") or "").strip()
|
||||
if not description:
|
||||
return jsonify({"error": "description cannot be empty"}), 400
|
||||
fields.append("description = ?")
|
||||
params.append(description)
|
||||
|
||||
if "jpy" in data:
|
||||
try:
|
||||
jpy_f = float(data.get("jpy"))
|
||||
except (TypeError, ValueError):
|
||||
return jsonify({"error": "jpy must be a number"}), 400
|
||||
fields.append("jpy = ?")
|
||||
params.append(jpy_f)
|
||||
|
||||
if "eur" in data:
|
||||
try:
|
||||
eur_f = float(data.get("eur"))
|
||||
except (TypeError, ValueError):
|
||||
return jsonify({"error": "eur must be a number"}), 400
|
||||
fields.append("eur = ?")
|
||||
params.append(eur_f)
|
||||
|
||||
if not fields:
|
||||
return jsonify({"error": "no fields to update"}), 400
|
||||
|
||||
params.append(expense_id)
|
||||
cur = g.db.execute(
|
||||
f"UPDATE expenses SET {', '.join(fields)} WHERE id = ?",
|
||||
tuple(params),
|
||||
)
|
||||
g.db.commit()
|
||||
if cur.rowcount == 0:
|
||||
return jsonify({"error": "not found"}), 404
|
||||
return jsonify({"ok": True})
|
||||
|
||||
@app.delete("/api/expenses/<int:expense_id>")
|
||||
def api_expenses_delete(expense_id: int):
|
||||
require_auth()
|
||||
cur = g.db.execute("DELETE FROM expenses WHERE id = ?", (expense_id,))
|
||||
g.db.commit()
|
||||
if cur.rowcount == 0:
|
||||
return jsonify({"error": "not found"}), 404
|
||||
return jsonify({"ok": True})
|
||||
|
||||
# -----------------
|
||||
# Errors
|
||||
# -----------------
|
||||
|
||||
@app.errorhandler(401)
|
||||
def _unauthorized(_e):
|
||||
return redirect(url_for("login"))
|
||||
|
||||
return app
|
||||
|
||||
|
||||
app = create_app()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
app.run(host="0.0.0.0", port=int(os.environ.get("PORT", "8000")), debug=False)
|
||||
Reference in New Issue
Block a user