feat(dev): 확정 없이 다음으로 — 개발 전용 잠금 해제 (신규 파일만, 1/2)

공용 파일 두 걸음 규칙 — 이 커밋은 신규 파일뿐이고 main.py 등록은 다음 커밋. 먼저 push 해야 다른 창이 등록 줄만 받고 파일이 없어 죽는 일이 없음. 까닭 — 단계마다 [확정]을 해야 다음 페이지가 열려서, 상세 설계를 확정하기 전에는 B08·B09 를 볼 수 없음. 오늘 화면 검증이 그 자리에서 두 번 막혔음. ⚠ 계산을 대신 돌리지 않음 — 잠금만 품 - 바꾸는 것은 project_workflow_stages.state 한 칸뿐 - 확정은 전 측점을 다시 계산해 정본에 쓰는 것이라, 흉내 내면 「확정 안 했는데 확정된 값」이 생겨 막힌 것보다 나쁨 - 값이 없으면 B08 이 「미확보」로 뜨는 것이 정상 ⚠ 문은 서버가 정본 — ENVIRONMENT 가 개발이 아니면 세 입구 모두 403 (화면에서 단추를 숨겨도 API 가 열려 있으면 소용없음) 되돌리기 — 푼 단계의 이전 상태를 message 에 DEV_UNLOCK:<옛상태> 로 적어 두고 DELETE 가 그대로 복원. 그 표시가 없는 단계(진짜 확정)는 안 건드림. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-08 06:39:05 +09:00
parent de31cd9c42
commit 618b6cf4c5
2 changed files with 278 additions and 0 deletions
@@ -0,0 +1,113 @@
+"""개발환경 전용 — 「확정 없이 다음으로」 API.
+
+⚠ **문은 여기가 정본이다.** 화면에서 단추를 숨겨도 API 가 열려 있으면 아무 소용이 없다.
+   그래서 세 입구 모두 `require_dev_environment()` 를 먼저 부르고, 운영에서는 **403** 으로
+   거절한다. 프론트의 `import.meta.env.DEV` 는 보조일 뿐이다.
+
+⚠ **계산을 대신 돌리지 않는다.** 여기서 바뀌는 것은 `project_workflow_stages.state`
+   한 칸뿐이다. 자세한 까닭은 `common_util_dev_unlock` 의 머리말을 볼 것.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+from uuid import UUID
+
+import aiomysql
+from fastapi import APIRouter
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+
+from common_util.common_util_dev_unlock import (
+    DevUnlockDisabled,
+    relock_stages,
+    unlock_stages,
+    unlock_status,
+)
+from config.config_db import run_with_connection
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/projects", tags=["DEV Unlock"])
+
+#: 기본은 B08(수량산출)까지 — 검증이 가장 자주 막히던 자리다.
+DEFAULT_UP_TO_STAGE = 5
+
+
+class UnlockBody(BaseModel):
+    """어디까지 열 것인가. 안 주면 B08 까지."""
+
+    up_to_stage: int = DEFAULT_UP_TO_STAGE
+
+
+async def _with_cursor(connection: aiomysql.Connection, call: Any, *args: Any) -> Any:
+    """쓰기라 **한 커넥션·한 트랜잭션**으로 묶는다(`run_with_connection` 주석 참조)."""
+    async with connection.cursor(aiomysql.DictCursor) as cursor:
+        result = await call(cursor, *args)
+    await connection.commit()
+    return result
+
+
+@router.get("/{project_id}/dev/unlock")
+async def get_unlock_status(project_id: UUID) -> JSONResponse:
+    """지금 우회로 열려 있는 단계. **화면 안내의 근거**다."""
+
+    async def call(connection: aiomysql.Connection) -> dict[str, Any]:
+        async with connection.cursor(aiomysql.DictCursor) as cursor:
+            return await unlock_status(cursor, str(project_id))
+
+    try:
+        payload = await run_with_connection(call)
+    except Exception:
+        logger.exception("개발 우회 상태 조회 실패: project_id=%s", project_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "우회 상태를 읽지 못했습니다."},
+        )
+    return JSONResponse(content={"status": "success", **payload})
+
+
+@router.post("/{project_id}/dev/unlock")
+async def post_unlock(project_id: UUID, body: UnlockBody | None = None) -> JSONResponse:
+    """확정을 건너뛰고 **잠금만** 푼다. 개발환경이 아니면 403."""
+    up_to = (body or UnlockBody()).up_to_stage
+
+    async def call(connection: aiomysql.Connection) -> dict[str, Any]:
+        return await _with_cursor(connection, unlock_stages, str(project_id), up_to)
+
+    try:
+        payload = await run_with_connection(call)
+    except DevUnlockDisabled as error:
+        return JSONResponse(status_code=403, content={"status": "error", "message": str(error)})
+    except ValueError as error:
+        return JSONResponse(status_code=400, content={"status": "error", "message": str(error)})
+    except Exception:
+        logger.exception("개발 우회 실패: project_id=%s", project_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "단계를 열지 못했습니다."},
+        )
+    logger.info("DEV 우회: project_id=%s 까지=%s", project_id, up_to)
+    return JSONResponse(content={"status": "success", **payload})
+
+
+@router.delete("/{project_id}/dev/unlock")
+async def delete_unlock(project_id: UUID) -> JSONResponse:
+    """우회로 연 단계를 **원래 상태로 되돌린다.** 진짜 확정은 안 건드린다."""
+
+    async def call(connection: aiomysql.Connection) -> dict[str, Any]:
+        return await _with_cursor(connection, relock_stages, str(project_id))
+
+    try:
+        payload = await run_with_connection(call)
+    except DevUnlockDisabled as error:
+        return JSONResponse(status_code=403, content={"status": "error", "message": str(error)})
+    except Exception:
+        logger.exception("개발 우회 되돌리기 실패: project_id=%s", project_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "단계를 되돌리지 못했습니다."},
+        )
+    logger.info("DEV 우회 되돌림: project_id=%s", project_id)
+    return JSONResponse(content={"status": "success", **payload})