feat(cupang): 등록된 제품 수정 팝업

- 동작 열에 [수정] 버튼 → 제품명/제품코드/쿠팡상품코드 수정 팝업
- POST /cupang/api/products/{id}/edit (JSON), 제품코드 중복은 400
- 제품코드 변경 시 cupang_box_rules 의 product_code/스냅샷도 함께 이동
  (같은 코드 규칙이 이미 있으면 이동하지 않음). 출고 라인은 과거 기록이라 유지

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-01 12:00:33 +09:00
parent 9ddced268f
commit 7174edd702
3 changed files with 178 additions and 0 deletions
+67
View File
@@ -338,6 +338,73 @@ class CupangDBStore:
if cur.rowcount == 0:
raise KeyError(product_id)
def update_product(
self,
*,
product_id: int,
product_name: str,
product_code: str,
coupang_item_code: str = "",
) -> dict[str, Any]:
"""등록된 제품 수정(제품명/제품코드/쿠팡상품코드).
제품코드가 바뀌면 같은 코드로 연결돼 있던 박스 규칙(cupang_box_rules)의
product_code 도 함께 옮긴다(출고 라인은 과거 기록이라 스냅샷 유지).
"""
name = (product_name or "").strip()
code = (product_code or "").strip()
cic = (coupang_item_code or "").strip()
if not name or not code:
raise ValueError("제품코드와 제품명 모두 필요합니다.")
with self._pool.connection() as conn:
with conn.transaction():
old = conn.execute(
"SELECT * FROM cupang_products WHERE id = %s FOR UPDATE",
(product_id,),
).fetchone()
if old is None:
raise KeyError(product_id)
old_code = (old["product_code"] or "").strip()
if code != old_code:
dup = conn.execute(
"SELECT 1 FROM cupang_products "
"WHERE product_code = %s AND id <> %s",
(code, product_id),
).fetchone()
if dup:
raise ValueError(f"제품코드 {code} 는 이미 등록돼 있습니다.")
row = conn.execute(
"""
UPDATE cupang_products
SET product_name = %s,
product_code = %s,
coupang_item_code = %s
WHERE id = %s
RETURNING *
""",
(name, code, cic, product_id),
).fetchone()
if code != old_code:
# 새 코드로 된 규칙이 이미 있으면 옮기지 않는다(중복 방지).
exists_new = conn.execute(
"SELECT 1 FROM cupang_box_rules WHERE product_code = %s",
(code,),
).fetchone()
if not exists_new:
conn.execute(
"UPDATE cupang_box_rules "
"SET product_code = %s, product_name_snapshot = %s "
"WHERE product_code = %s",
(code, name, old_code),
)
else:
conn.execute(
"UPDATE cupang_box_rules SET product_name_snapshot = %s "
"WHERE product_code = %s",
(name, code),
)
return self._product_serialize(row)
def toggle_product_active(self, *, product_id: int) -> dict[str, Any]:
"""활성 ↔ 비활성 뒤집기. 갱신된 행을 반환."""
with self._pool.connection() as conn: