refactor: portfolio → personal 리네이밍 + Blog/Todo 통합
- portfolio/ 디렉토리를 personal/로 리네이밍 - lotto-backend의 Blog/Todo 라우트·CRUD를 personal 서비스로 이전 - lotto-backend에서 Blog/Todo 코드 제거 (DB 테이블 스키마는 유지) - nginx: /api/todos, /api/blog/ 라우팅을 personal로 추가 - docker-compose: portfolio → personal 서비스 변�� - deploy 스크립트: portfolio → personal 반영 데이터 마이그레이션은 배포 후 NAS에서 별도 수행 필요: 1. cp data/portfolio/portfolio.db data/personal/personal.db 2. sqlite3 data/lotto.db ".dump todos" | sqlite3 data/personal/personal.db 3. sqlite3 data/lotto.db ".dump blog_posts" | sqlite3 data/personal/personal.db Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -298,133 +298,6 @@ def init_db() -> None:
|
||||
conn.execute("CREATE INDEX IF NOT EXISTS idx_briefings_draw ON lotto_briefings(draw_no 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
|
||||
|
||||
|
||||
# ── 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:
|
||||
|
||||
@@ -15,10 +15,6 @@ from .db import (
|
||||
update_recommendation,
|
||||
# 시뮬레이션 관련
|
||||
get_best_picks, get_simulation_runs, get_simulation_candidates,
|
||||
# todos
|
||||
get_all_todos, create_todo, update_todo, delete_todo, delete_done_todos,
|
||||
# blog
|
||||
get_all_posts, create_post, update_post, delete_post,
|
||||
# 성과 통계
|
||||
get_recommendation_performance,
|
||||
# Phase 2: 구매 이력
|
||||
@@ -839,99 +835,3 @@ def version():
|
||||
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}
|
||||
|
||||
|
||||
# ── Blog API ──────────────────────────────────────────────────────────────────
|
||||
|
||||
class BlogPostCreate(BaseModel):
|
||||
title: str
|
||||
body: str = ""
|
||||
excerpt: str = ""
|
||||
tags: List[str] = []
|
||||
date: str = "" # 빈 문자열이면 오늘 날짜 사용
|
||||
|
||||
|
||||
class BlogPostUpdate(BaseModel):
|
||||
title: Optional[str] = None
|
||||
body: Optional[str] = None
|
||||
excerpt: Optional[str] = None
|
||||
tags: Optional[List[str]] = None
|
||||
date: Optional[str] = None
|
||||
|
||||
|
||||
@app.get("/api/blog/posts")
|
||||
def api_blog_list():
|
||||
return {"posts": get_all_posts()}
|
||||
|
||||
|
||||
@app.post("/api/blog/posts", status_code=201)
|
||||
def api_blog_create(body: BlogPostCreate):
|
||||
from datetime import date as _date
|
||||
post_date = body.date if body.date else _date.today().isoformat()
|
||||
post = create_post(body.title, body.body, body.excerpt, body.tags, post_date)
|
||||
return post
|
||||
|
||||
|
||||
@app.put("/api/blog/posts/{post_id}")
|
||||
def api_blog_update(post_id: int, body: BlogPostUpdate):
|
||||
updated = update_post(post_id, body.model_dump(exclude_none=True))
|
||||
if updated is None:
|
||||
raise HTTPException(status_code=404, detail="Post not found")
|
||||
return updated
|
||||
|
||||
|
||||
@app.delete("/api/blog/posts/{post_id}")
|
||||
def api_blog_delete(post_id: int):
|
||||
ok = delete_post(post_id)
|
||||
if not ok:
|
||||
raise HTTPException(status_code=404, detail="Post not found")
|
||||
return {"ok": True}
|
||||
|
||||
Reference in New Issue
Block a user