king
cb43fe1257
- ItemcodeReader.last_error 추가, _run 실패 시 logger.exception + 메시지 보존 - /api/products/all 응답에 error/count 추가 - products 화면: 결과 0 일 때 조회오류/사유 표시 (그냥 '결과 없음' 대신) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
156 lines
6.4 KiB
Python
156 lines
6.4 KiB
Python
r"""itemcode_db 읽기 전용 상품 검색.
|
|
|
|
cupang 모듈은 itemcode_db 의 상품(낱개코드/세트코드)을 **읽기만** 한다.
|
|
cupang_db 에 상품을 복제 저장하지 않는다. 출고 라인에는 product_code 와
|
|
product_name_snapshot 만 보존한다.
|
|
|
|
⚠️ itemcode_db 의 실제 테이블/컬럼명은 이 저장소(main-app)에 정의되어 있지 않다.
|
|
운영 서버에서 다음으로 먼저 구조를 확인한 뒤 환경변수를 설정해야 한다:
|
|
|
|
docker exec -it postgres-db psql -U postgres -d itemcode_db -c "\dt"
|
|
docker exec -it postgres-db psql -U postgres -d itemcode_db -c "\d <테이블명>"
|
|
|
|
환경변수 (모두 미설정 시 검색 비활성 → 폼에서 수동 입력으로 폴백):
|
|
|
|
ITEMCODE_DB_URL 읽기 전용 DSN. 예: postgresql://itemcode_ro:<pwd>@postgres-db:5432/itemcode_db
|
|
ITEMCODE_SEARCH_SQL (선택) 검색 SQL 직접 지정. 아래 자동 생성 대신 사용.
|
|
반드시 code, name, type 컬럼을 별칭으로 반환하고,
|
|
%(q)s 파라미터를 LIKE 패턴으로 받는다.
|
|
|
|
자동 생성용 (ITEMCODE_SEARCH_SQL 미설정 시):
|
|
ITEMCODE_TABLE 검색 대상 테이블/뷰 (예: products 또는 item_master)
|
|
ITEMCODE_CODE_COL 코드 컬럼명 (기본: product_code)
|
|
ITEMCODE_NAME_COL 상품명 컬럼명 (기본: product_name)
|
|
ITEMCODE_TYPE_COL (선택) 단품/세트 구분 컬럼명. 없으면 type 은 빈 문자열.
|
|
|
|
낱개코드와 세트코드가 별도 테이블이면 ITEMCODE_SEARCH_SQL 에 UNION 으로 직접 작성한다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger("cupang.itemcode")
|
|
|
|
# 안전한 SQL 식별자(테이블/컬럼)만 허용 — 인젝션 방지.
|
|
_IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_.]*$")
|
|
|
|
|
|
def _ident(value: str, *, what: str) -> str:
|
|
v = (value or "").strip()
|
|
if not _IDENT_RE.match(v):
|
|
raise ValueError(f"안전하지 않은 {what} 식별자: {value!r}")
|
|
return v
|
|
|
|
|
|
class ItemcodeReader:
|
|
"""itemcode_db 읽기 전용 커넥션 풀 + 상품 검색.
|
|
|
|
설정이 없거나 불완전하면 `enabled=False` 로 두고, search()는 빈 리스트를 반환한다.
|
|
앱 부팅이나 cupang 모듈 진입을 막지 않는다.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._pool: Any = None
|
|
self._sql: str | None = None
|
|
self.enabled = False
|
|
self.reason = ""
|
|
self.last_error = "" # 마지막 조회 오류(진단용, 비밀값 없음)
|
|
self._configure()
|
|
|
|
def _configure(self) -> None:
|
|
dsn = os.getenv("ITEMCODE_DB_URL", "").strip()
|
|
if not dsn:
|
|
self.reason = "ITEMCODE_DB_URL 미설정 — 상품 검색 비활성(수동 입력 사용)."
|
|
return
|
|
|
|
custom_sql = os.getenv("ITEMCODE_SEARCH_SQL", "").strip()
|
|
if custom_sql:
|
|
self._sql = custom_sql
|
|
else:
|
|
table = os.getenv("ITEMCODE_TABLE", "").strip()
|
|
if not table:
|
|
self.reason = (
|
|
"ITEMCODE_TABLE(또는 ITEMCODE_SEARCH_SQL) 미설정 — "
|
|
"상품 검색 비활성(수동 입력 사용)."
|
|
)
|
|
return
|
|
try:
|
|
table_id = _ident(table, what="테이블")
|
|
code_col = _ident(os.getenv("ITEMCODE_CODE_COL", "product_code"), what="코드 컬럼")
|
|
name_col = _ident(os.getenv("ITEMCODE_NAME_COL", "product_name"), what="상품명 컬럼")
|
|
type_col_raw = os.getenv("ITEMCODE_TYPE_COL", "").strip()
|
|
type_expr = _ident(type_col_raw, what="구분 컬럼") if type_col_raw else "''"
|
|
except ValueError as exc:
|
|
self.reason = f"itemcode 검색 설정 오류: {exc}"
|
|
return
|
|
self._sql = (
|
|
f"SELECT {code_col} AS code, {name_col} AS name, {type_expr} AS type "
|
|
f"FROM {table_id} "
|
|
f"WHERE {code_col} ILIKE %(q)s OR {name_col} ILIKE %(q)s "
|
|
f"ORDER BY {code_col} ASC LIMIT %(limit)s"
|
|
)
|
|
|
|
# 풀은 lazy open — 부팅 시 itemcode_db 가 잠시 끊겨도 죽지 않게.
|
|
try:
|
|
from psycopg.rows import dict_row
|
|
from psycopg_pool import ConnectionPool
|
|
|
|
self._pool = ConnectionPool(
|
|
conninfo=dsn,
|
|
min_size=1,
|
|
max_size=3,
|
|
kwargs={"row_factory": dict_row, "autocommit": True},
|
|
open=False,
|
|
)
|
|
self._pool.open(wait=False)
|
|
self.enabled = True
|
|
self.reason = ""
|
|
except Exception as exc: # noqa: BLE001 — 설정/드라이버 문제로 모듈을 죽이지 않음
|
|
self.reason = f"itemcode_db 연결 풀 생성 실패: {type(exc).__name__}"
|
|
|
|
def search(self, query: str, *, limit: int = 20) -> list[dict[str, Any]]:
|
|
"""code/name 부분 일치 검색. 반환: [{"code","name","type"}].
|
|
|
|
비활성 상태이거나 조회 실패 시 빈 리스트(예외 비전파 — UI 는 수동 입력 폴백).
|
|
"""
|
|
q = (query or "").strip()
|
|
if not q:
|
|
return []
|
|
return self._run(f"%{q}%", limit)
|
|
|
|
def list_all(self, *, limit: int = 2000) -> list[dict[str, Any]]:
|
|
"""전체 상품 목록(낱개+세트). 설정 화면 왼쪽 리스트용."""
|
|
return self._run("%", limit)
|
|
|
|
def _run(self, like: str, limit: int) -> list[dict[str, Any]]:
|
|
if not self.enabled or not self._pool or not self._sql:
|
|
return []
|
|
try:
|
|
with self._pool.connection() as conn:
|
|
rows = conn.execute(
|
|
self._sql, {"q": like, "limit": int(limit)}
|
|
).fetchall()
|
|
self.last_error = ""
|
|
except Exception as exc: # noqa: BLE001 — 모듈을 죽이지 않음. 원인은 로그 + last_error.
|
|
self.last_error = f"{type(exc).__name__}: {exc}"
|
|
logger.exception("itemcode 조회 실패 (SQL/스키마 확인 필요)")
|
|
return []
|
|
out: list[dict[str, Any]] = []
|
|
for r in rows:
|
|
out.append(
|
|
{
|
|
"code": str(r.get("code") or "").strip(),
|
|
"name": str(r.get("name") or "").strip(),
|
|
"type": str(r.get("type") or "").strip(),
|
|
}
|
|
)
|
|
return out
|
|
|
|
def close(self) -> None:
|
|
if self._pool is not None:
|
|
self._pool.close()
|