eomsangdonandClaude Opus 5
618b6cf4c5
공용 파일 두 걸음 규칙 — 이 커밋은 신규 파일뿐이고 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>
114 lines
4.5 KiB
Python
114 lines
4.5 KiB
Python
"""개발환경 전용 — 「확정 없이 다음으로」 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})
|