app.py

import os
from datetime import datetime
from flask import Flask, request, jsonify
from sqlalchemy import (
    create_engine, String, Integer, Boolean, DateTime, Text
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker
from sqlalchemy.exc import SQLAlchemyError
from dotenv import load_dotenv

load_dotenv()

def env(name: str, default: str | None = None) -> str:
    v = os.getenv(name, default)
    if v is None or v == "":
        raise RuntimeError(f"Missing env var: {name}")
    return v

APP_HOST = os.getenv("APP_HOST", "0.0.0.0")
APP_PORT = int(os.getenv("APP_PORT", "4000"))
APP_DEBUG = os.getenv("APP_DEBUG", "false").lower() == "true"

DB_HOST = env("DB_HOST")
DB_PORT = env("DB_PORT", "5432")
DB_NAME = env("DB_NAME")
DB_USER = env("DB_USER")
DB_PASSWORD = env("DB_PASSWORD")

DATABASE_URL = (
    f"postgresql+psycopg2://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
)

engine = create_engine(
    DATABASE_URL,
    pool_pre_ping=True,
    pool_size=5,
    max_overflow=10,
    future=True,
)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)

class Base(DeclarativeBase):
    pass

class Todo(Base):
    __tablename__ = "todos"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    description: Mapped[str | None] = mapped_column(Text, nullable=True)
    done: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=datetime.utcnow)

def init_db() -> None:
    Base.metadata.create_all(bind=engine)

app = Flask(__name__)
init_db()

@app.get("/health")
def health():
    try:
        with engine.connect() as conn:
            conn.exec_driver_sql("SELECT 1;")
        return jsonify({"status": "ok"}), 200
    except Exception as e:
        return jsonify({"status": "db_error", "error": str(e)}), 500

@app.get("/todos")
def list_todos():
    with SessionLocal() as db:
        todos = db.query(Todo).order_by(Todo.id.desc()).all()
        return jsonify([
            {
                "id": t.id,
                "title": t.title,
                "description": t.description,
                "done": t.done,
                "created_at": t.created_at.isoformat() + "Z",
            }
            for t in todos
        ])

@app.post("/todos")
def create_todo():
    data = request.get_json(silent=True) or {}
    title = (data.get("title") or "").strip()
    if not title:
        return jsonify({"error": "title is required"}), 400

    description = (data.get("description") or "").strip() or None

    with SessionLocal() as db:
        try:
            todo = Todo(title=title, description=description, done=False)
            db.add(todo)
            db.commit()
            db.refresh(todo)
            return jsonify({"id": todo.id, "title": todo.title, "description": todo.description, "done": todo.done}), 201
        except SQLAlchemyError as e:
            db.rollback()
            return jsonify({"error": "db_error", "detail": str(e)}), 500

@app.patch("/todos/<int:todo_id>")
def update_todo(todo_id: int):
    data = request.get_json(silent=True) or {}
    with SessionLocal() as db:
        todo = db.get(Todo, todo_id)
        if not todo:
            return jsonify({"error": "not_found"}), 404

        if "title" in data:
            new_title = (data.get("title") or "").strip()
            if not new_title:
                return jsonify({"error": "title cannot be empty"}), 400
            todo.title = new_title

        if "description" in data:
            desc = (data.get("description") or "").strip()
            todo.description = desc or None

        if "done" in data:
            todo.done = bool(data.get("done"))

        try:
            db.commit()
            return jsonify({"id": todo.id, "title": todo.title, "description": todo.description, "done": todo.done}), 200
        except SQLAlchemyError as e:
            db.rollback()
            return jsonify({"error": "db_error", "detail": str(e)}), 500

@app.delete("/todos/<int:todo_id>")
def delete_todo(todo_id: int):
    with SessionLocal() as db:
        todo = db.get(Todo, todo_id)
        if not todo:
            return jsonify({"error": "not_found"}), 404
        try:
            db.delete(todo)
            db.commit()
            return "", 204
        except SQLAlchemyError as e:
            db.rollback()
            return jsonify({"error": "db_error", "detail": str(e)}), 500

if __name__ == "__main__":
    app.run(host=APP_HOST, port=APP_PORT, debug=APP_DEBUG)