블로그 글 작성 api 추가

backend/app/db.py

@@ -162,6 +162,25 @@ def init_db() -> None:
             "CREATE INDEX IF NOT EXISTS idx_todos_created ON todos(created_at DESC);"
         )
 
+        # ── blog_posts 테이블 ──────────────────────────────────────────────────
+        conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS blog_posts (
+                id         INTEGER PRIMARY KEY AUTOINCREMENT,
+                title      TEXT    NOT NULL,
+                body       TEXT    NOT NULL DEFAULT '',
+                excerpt    TEXT    NOT NULL DEFAULT '',
+                tags       TEXT    NOT NULL DEFAULT '[]',
+                date       TEXT    NOT NULL DEFAULT (date('now','localtime')),
+                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_blog_date ON blog_posts(date DESC);"
+        )
+
 
 # ── todos CRUD ───────────────────────────────────────────────────────────────
 
@@ -230,6 +249,68 @@ def delete_done_todos() -> int:
     return cur.rowcount
 
 
+# ── blog_posts CRUD ──────────────────────────────────────────────────────────
+
+def _post_row_to_dict(r) -> Dict[str, Any]:
+    return {
+        "id": r["id"],
+        "title": r["title"],
+        "body": r["body"],
+        "excerpt": r["excerpt"],
+        "tags": json.loads(r["tags"]) if r["tags"] else [],
+        "date": r["date"],
+        "created_at": r["created_at"],
+        "updated_at": r["updated_at"],
+    }
+
+
+def get_all_posts() -> List[Dict[str, Any]]:
+    with _conn() as conn:
+        rows = conn.execute(
+            "SELECT * FROM blog_posts ORDER BY date DESC, id DESC"
+        ).fetchall()
+    return [_post_row_to_dict(r) for r in rows]
+
+
+def create_post(title: str, body: str, excerpt: str, tags: List[str], date: str) -> Dict[str, Any]:
+    with _conn() as conn:
+        conn.execute(
+            "INSERT INTO blog_posts (title, body, excerpt, tags, date) VALUES (?, ?, ?, ?, ?)",
+            (title, body, excerpt, json.dumps(tags), date),
+        )
+        row = conn.execute(
+            "SELECT * FROM blog_posts WHERE rowid = last_insert_rowid()"
+        ).fetchone()
+    return _post_row_to_dict(row)
+
+
+def update_post(post_id: int, fields: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+    allowed = {"title", "body", "excerpt", "tags", "date"}
+    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 blog_posts WHERE id = ?", (post_id,)).fetchone()
+        return _post_row_to_dict(row) if row else None
+
+    if "tags" in updates:
+        updates["tags"] = json.dumps(updates["tags"])
+
+    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()) + [post_id]
+
+    with _conn() as conn:
+        conn.execute(f"UPDATE blog_posts SET {set_clauses} WHERE id = ?", args)
+        row = conn.execute("SELECT * FROM blog_posts WHERE id = ?", (post_id,)).fetchone()
+    return _post_row_to_dict(row) if row else None
+
+
+def delete_post(post_id: int) -> bool:
+    with _conn() as conn:
+        cur = conn.execute("DELETE FROM blog_posts WHERE id = ?", (post_id,))
+    return cur.rowcount > 0
+
+
 def upsert_draw(row: Dict[str, Any]) -> None:
     with _conn() as conn:
         conn.execute(