"""TaskFlow — Flask task management application.""" import os import psycopg2 import psycopg2.extras from flask import Flask, request, redirect, url_for, render_template_string, jsonify app = Flask(__name__) def get_db(): return psycopg2.connect( host=os.environ.get("POSTGRES_HOST"), port=int(os.environ.get("POSTGRES_PORT", 5432)), user=os.environ["POSTGRES_USER"], password=os.environ["POSTGRES_PASSWORD"], dbname=os.environ["POSTGRES_DB"], sslmode="require" ) @app.route("/health") def health(): try: conn = get_db(); conn.close() return jsonify({"status": "ok"}), 200 except Exception as e: return jsonify({"status": "error", "detail": str(e)}), 500 TEMPLATE = """ TaskFlow
To Do {{tasks['todo']|length}} In Progress {{tasks['in_progress']|length}} Done {{tasks['done']|length}}
New Task
{% for col,label,css,icon in [ ('todo','To Do','col-todo','◯'), ('in_progress','In Progress','col-progress','▶'), ('done','Done','col-done','✓')] %}
{{icon}} {{label}}
{{tasks[col]|length}}
{% if tasks[col] %} {% for t in tasks[col] %}
{{t.title}}
{% if t.description %}
{{t.description}}
{% endif %}
{% if col != 'done' %}
{% endif %}
{{t.created_at.strftime('%b %d') if t.created_at else ''}}
{% endfor %} {% else %}
▫
Nothing here yet
{% endif %}
{% endfor %}
""" NEXT_STATUS = {"todo": "in_progress", "in_progress": "done"} @app.route("/") def index(): conn = get_db() cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) cur.execute("SELECT * FROM tasks ORDER BY created_at") rows = cur.fetchall() cur.close(); conn.close() tasks = {"todo": [], "in_progress": [], "done": []} for row in rows: tasks[row["status"]].append(row) return render_template_string(TEMPLATE, tasks=tasks) @app.route("/create", methods=["POST"]) def create(): title = request.form.get("title", "").strip() desc = request.form.get("description", "").strip() if title: conn = get_db(); cur = conn.cursor() cur.execute( "INSERT INTO tasks (title, description, status) VALUES (%s, %s, 'todo')", (title, desc or None)) conn.commit(); cur.close(); conn.close() return redirect(url_for("index")) @app.route("/advance/", methods=["POST"]) def advance(task_id): conn = get_db(); cur = conn.cursor() cur.execute("SELECT status FROM tasks WHERE id = %s", (task_id,)) row = cur.fetchone() if row and row[0] in NEXT_STATUS: cur.execute("UPDATE tasks SET status = %s WHERE id = %s", (NEXT_STATUS[row[0]], task_id)) conn.commit() cur.close(); conn.close() return redirect(url_for("index")) @app.route("/delete/", methods=["POST"]) def delete(task_id): conn = get_db(); cur = conn.cursor() cur.execute("DELETE FROM tasks WHERE id = %s", (task_id,)) conn.commit(); cur.close(); conn.close() return redirect(url_for("index")) if __name__ == "__main__": app.run(host="0.0.0.0", port=5000, debug=False)