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/") 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/") 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/") 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/") 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)