todo List 작성 api 추가
This commit is contained in:
@@ -143,6 +143,93 @@ def init_db() -> None:
|
|||||||
"ON best_picks(is_active, score_total DESC);"
|
"ON best_picks(is_active, score_total DESC);"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# ── todos 테이블 ───────────────────────────────────────────────────────
|
||||||
|
conn.execute(
|
||||||
|
"""
|
||||||
|
CREATE TABLE IF NOT EXISTS todos (
|
||||||
|
id TEXT PRIMARY KEY
|
||||||
|
DEFAULT (lower(hex(randomblob(4))) || '-' || lower(hex(randomblob(2)))),
|
||||||
|
title TEXT NOT NULL,
|
||||||
|
description TEXT,
|
||||||
|
status TEXT NOT NULL DEFAULT 'todo'
|
||||||
|
CHECK(status IN ('todo','in_progress','done')),
|
||||||
|
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
|
||||||
|
updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
|
||||||
|
);
|
||||||
|
"""
|
||||||
|
)
|
||||||
|
conn.execute(
|
||||||
|
"CREATE INDEX IF NOT EXISTS idx_todos_created ON todos(created_at DESC);"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# ── todos CRUD ───────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
def _todo_row_to_dict(r) -> Dict[str, Any]:
|
||||||
|
return {
|
||||||
|
"id": r["id"],
|
||||||
|
"title": r["title"],
|
||||||
|
"description": r["description"],
|
||||||
|
"status": r["status"],
|
||||||
|
"created_at": r["created_at"],
|
||||||
|
"updated_at": r["updated_at"],
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def get_all_todos() -> List[Dict[str, Any]]:
|
||||||
|
with _conn() as conn:
|
||||||
|
rows = conn.execute(
|
||||||
|
"SELECT * FROM todos ORDER BY created_at DESC"
|
||||||
|
).fetchall()
|
||||||
|
return [_todo_row_to_dict(r) for r in rows]
|
||||||
|
|
||||||
|
|
||||||
|
def create_todo(title: str, description: Optional[str], status: str) -> Dict[str, Any]:
|
||||||
|
with _conn() as conn:
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO todos (title, description, status) VALUES (?, ?, ?)",
|
||||||
|
(title, description, status),
|
||||||
|
)
|
||||||
|
row = conn.execute(
|
||||||
|
"SELECT * FROM todos WHERE rowid = last_insert_rowid()"
|
||||||
|
).fetchone()
|
||||||
|
return _todo_row_to_dict(row)
|
||||||
|
|
||||||
|
|
||||||
|
def update_todo(todo_id: str, fields: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
||||||
|
"""fields에 있는 항목만 업데이트 (PATCH 방식), updated_at 자동 갱신"""
|
||||||
|
allowed = {"title", "description", "status"}
|
||||||
|
updates = {k: v for k, v in fields.items() if k in allowed}
|
||||||
|
if not updates:
|
||||||
|
with _conn() as conn:
|
||||||
|
row = conn.execute("SELECT * FROM todos WHERE id = ?", (todo_id,)).fetchone()
|
||||||
|
return _todo_row_to_dict(row) if row else None
|
||||||
|
|
||||||
|
set_clauses = ", ".join(f"{k} = ?" for k in updates)
|
||||||
|
set_clauses += ", updated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now')"
|
||||||
|
args = list(updates.values()) + [todo_id]
|
||||||
|
|
||||||
|
with _conn() as conn:
|
||||||
|
conn.execute(
|
||||||
|
f"UPDATE todos SET {set_clauses} WHERE id = ?",
|
||||||
|
args,
|
||||||
|
)
|
||||||
|
row = conn.execute("SELECT * FROM todos WHERE id = ?", (todo_id,)).fetchone()
|
||||||
|
return _todo_row_to_dict(row) if row else None
|
||||||
|
|
||||||
|
|
||||||
|
def delete_todo(todo_id: str) -> bool:
|
||||||
|
with _conn() as conn:
|
||||||
|
cur = conn.execute("DELETE FROM todos WHERE id = ?", (todo_id,))
|
||||||
|
return cur.rowcount > 0
|
||||||
|
|
||||||
|
|
||||||
|
def delete_done_todos() -> int:
|
||||||
|
with _conn() as conn:
|
||||||
|
cur = conn.execute("DELETE FROM todos WHERE status = 'done'")
|
||||||
|
return cur.rowcount
|
||||||
|
|
||||||
|
|
||||||
def upsert_draw(row: Dict[str, Any]) -> None:
|
def upsert_draw(row: Dict[str, Any]) -> None:
|
||||||
with _conn() as conn:
|
with _conn() as conn:
|
||||||
conn.execute(
|
conn.execute(
|
||||||
|
|||||||
@@ -10,6 +10,8 @@ from .db import (
|
|||||||
update_recommendation,
|
update_recommendation,
|
||||||
# 시뮬레이션 관련
|
# 시뮬레이션 관련
|
||||||
get_best_picks, get_simulation_runs, get_simulation_candidates,
|
get_best_picks, get_simulation_runs, get_simulation_candidates,
|
||||||
|
# todos
|
||||||
|
get_all_todos, create_todo, update_todo, delete_todo, delete_done_todos,
|
||||||
)
|
)
|
||||||
from .recommender import recommend_numbers, recommend_with_heatmap
|
from .recommender import recommend_numbers, recommend_with_heatmap
|
||||||
from .collector import sync_latest, sync_ensure_all
|
from .collector import sync_latest, sync_ensure_all
|
||||||
@@ -568,3 +570,54 @@ def api_recommend_batch_save(body: BatchSave):
|
|||||||
@app.get("/api/version")
|
@app.get("/api/version")
|
||||||
def version():
|
def version():
|
||||||
return {"version": os.getenv("APP_VERSION", "dev")}
|
return {"version": os.getenv("APP_VERSION", "dev")}
|
||||||
|
|
||||||
|
|
||||||
|
# ── Todos API ─────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
class TodoCreate(BaseModel):
|
||||||
|
title: str
|
||||||
|
description: Optional[str] = None
|
||||||
|
status: str = "todo"
|
||||||
|
|
||||||
|
|
||||||
|
class TodoUpdate(BaseModel):
|
||||||
|
title: Optional[str] = None
|
||||||
|
description: Optional[str] = None
|
||||||
|
status: Optional[str] = None
|
||||||
|
|
||||||
|
|
||||||
|
@app.get("/api/todos")
|
||||||
|
def api_todos_list():
|
||||||
|
return get_all_todos()
|
||||||
|
|
||||||
|
|
||||||
|
@app.post("/api/todos", status_code=201)
|
||||||
|
def api_todos_create(body: TodoCreate):
|
||||||
|
if body.status not in ("todo", "in_progress", "done"):
|
||||||
|
raise HTTPException(status_code=422, detail="status must be todo | in_progress | done")
|
||||||
|
return create_todo(body.title, body.description, body.status)
|
||||||
|
|
||||||
|
|
||||||
|
# ⚠️ /done 라우트를 /{todo_id} 보다 먼저 등록해야 done이 id로 매칭되지 않음
|
||||||
|
@app.delete("/api/todos/done")
|
||||||
|
def api_todos_delete_done():
|
||||||
|
deleted = delete_done_todos()
|
||||||
|
return {"deleted": deleted}
|
||||||
|
|
||||||
|
|
||||||
|
@app.put("/api/todos/{todo_id}")
|
||||||
|
def api_todos_update(todo_id: str, body: TodoUpdate):
|
||||||
|
if body.status is not None and body.status not in ("todo", "in_progress", "done"):
|
||||||
|
raise HTTPException(status_code=422, detail="status must be todo | in_progress | done")
|
||||||
|
updated = update_todo(todo_id, body.model_dump(exclude_none=True))
|
||||||
|
if updated is None:
|
||||||
|
raise HTTPException(status_code=404, detail="Todo not found")
|
||||||
|
return updated
|
||||||
|
|
||||||
|
|
||||||
|
@app.delete("/api/todos/{todo_id}")
|
||||||
|
def api_todos_delete(todo_id: str):
|
||||||
|
ok = delete_todo(todo_id)
|
||||||
|
if not ok:
|
||||||
|
raise HTTPException(status_code=404, detail="Todo not found")
|
||||||
|
return {"ok": True}
|
||||||
|
|||||||
Reference in New Issue
Block a user