feat(cafe24): 상품 상세페이지 관리 모듈 Phase 1
카페24 관리자에 직접 접속하지 않고 상품 상세페이지(description HTML)를 편집·예약 적용·복원하기 위한 모듈의 기반을 만든다. Phase 1 은 공통 Integration 계층, cafe24_db, OAuth 연결 화면까지다. 카페24 OAuth/API 클라이언트를 상품관리 모듈 안에 두지 않고 app/integrations/cafe24/ 로 분리했다. 향후 추가할 주문관리(주문 조회·송장 일괄등록·취소/반품/교환)가 같은 토큰과 클라이언트를 그대로 재사용해야 하기 때문이다. 라우터에서 httpx 를 직접 부르지 않고 Cafe24Client 만 쓰게 해서 재시도·rate limit·API 로그·토큰 갱신을 한 곳에 모았다. 토큰은 Fernet 으로 암호화해 저장한다(CAFE24_TOKEN_SECRET). DB 덤프가 유출돼도 access/refresh token 이 평문으로 남지 않게 하기 위함이며, API 로그와 연결 상태 화면에는 토큰·시크릿을 일절 기록/표시하지 않는다. 토큰 갱신은 행 잠금(SELECT ... FOR UPDATE) 안에서 한다. 카페24는 refresh token 을 회전시키므로, 이후 추가될 예약 worker 컨테이너와 web 컨테이너가 동시에 갱신하면 한쪽 토큰이 무효화된다. 기존 파일 변경은 목록에 한 줄씩 추가하는 형태로 44줄뿐이며 기존 라우트· 테이블·인증 로직은 건드리지 않았다. CAFE24_DB_URL 미설정 시 store 가 None 이라 앱은 정상 기동하고 모듈만 "설정 필요" 안내를 표시한다. 가드 헬퍼를 common.py 로 분리한 것은 router.py 가 routes_system.py 를 include 하는 구조에서 순환 import 가 생기기 때문이다. 검증: 신규 테스트 16개 통과(암호화 왕복, 토큰 만료·자동갱신, 상태 노출 시 토큰 미유출, 재시도 예산, 예약 상태 전이). dispatch 기존 테스트 9개 통과. cafe24_db_init.sql 은 로컬에 Docker 가 없어 미실행 — 서버 적용 시 확인 필요. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,237 @@
|
||||
"""cafe24_db PostgreSQL 저장소.
|
||||
|
||||
- 드라이버: psycopg 3 (`psycopg[binary,pool]`) — 다른 모듈과 동일 패턴.
|
||||
- 연결 정보: 환경변수 `CAFE24_DB_URL`
|
||||
(예: postgresql://cafe24_app:<pwd>@postgres-db:5432/cafe24_db)
|
||||
- 스키마는 앱이 만들지 않는다. `scripts/sql/cafe24_db_init.sql` 을 superuser 가
|
||||
사전 적용한다. 앱 계정(cafe24_app)은 CRUD 권한만 받는다.
|
||||
- 연결 풀은 lazy open — 부팅 시 DB 가 잠시 끊겨도 컨테이너가 죽지 않게.
|
||||
|
||||
토큰 값은 이 계층에 도달하기 전 이미 Fernet 암호문이다(평문 취급 금지).
|
||||
API 로그에는 토큰/시크릿을 넣지 않는다.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from contextlib import contextmanager
|
||||
from datetime import date, datetime
|
||||
from typing import Any, Iterator
|
||||
|
||||
from psycopg.rows import dict_row
|
||||
from psycopg_pool import ConnectionPool
|
||||
|
||||
from app.timezone import KST
|
||||
|
||||
from . import store
|
||||
|
||||
logger = logging.getLogger("cafe24.db")
|
||||
|
||||
# save_token_row / TokenLock.save 에서 부분 갱신을 허용하는 컬럼 화이트리스트.
|
||||
# 여기 없는 키는 무시한다(임의 컬럼 주입 방지).
|
||||
_TOKEN_FIELDS: tuple[str, ...] = (
|
||||
"access_token",
|
||||
"refresh_token",
|
||||
"access_token_expires_at",
|
||||
"refresh_token_expires_at",
|
||||
"scopes",
|
||||
"last_refreshed_at",
|
||||
"last_error",
|
||||
"connected_by",
|
||||
)
|
||||
|
||||
|
||||
class TokenLock:
|
||||
"""token_lock() 이 넘겨주는 핸들. 잠긴 행 조회 + 같은 트랜잭션 안 저장."""
|
||||
|
||||
def __init__(self, conn: Any, mall_id: str, row: dict[str, Any] | None):
|
||||
self._conn = conn
|
||||
self._mall_id = mall_id
|
||||
self.row = row
|
||||
|
||||
def save(self, **fields: Any) -> None:
|
||||
_update_token_row(self._conn, self._mall_id, fields)
|
||||
|
||||
|
||||
def _update_token_row(conn: Any, mall_id: str, fields: dict[str, Any]) -> None:
|
||||
"""UPSERT. 주어진 컬럼만 갱신한다(부분 갱신)."""
|
||||
allowed = {k: v for k, v in fields.items() if k in _TOKEN_FIELDS}
|
||||
if not allowed:
|
||||
return
|
||||
columns = list(allowed.keys())
|
||||
placeholders = ", ".join(["%s"] * len(columns))
|
||||
assignments = ", ".join(f"{col} = EXCLUDED.{col}" for col in columns)
|
||||
conn.execute(
|
||||
f"""
|
||||
INSERT INTO cafe24_oauth_tokens (mall_id, {", ".join(columns)})
|
||||
VALUES (%s, {placeholders})
|
||||
ON CONFLICT (mall_id) DO UPDATE SET {assignments}
|
||||
""",
|
||||
(mall_id, *[allowed[col] for col in columns]),
|
||||
)
|
||||
|
||||
|
||||
class Cafe24Store:
|
||||
def __init__(self, dsn: str, *, min_size: int = 1, max_size: int = 5):
|
||||
self._pool = ConnectionPool(
|
||||
conninfo=dsn,
|
||||
min_size=min_size,
|
||||
max_size=max_size,
|
||||
kwargs={"row_factory": dict_row, "autocommit": True},
|
||||
open=False,
|
||||
)
|
||||
self._pool.open(wait=False)
|
||||
|
||||
def close(self) -> None:
|
||||
self._pool.close()
|
||||
|
||||
# ════════════════════════════════════════════════════════════
|
||||
# OAuth 토큰 — app/integrations/cafe24/tokens.py 가 요구하는 3개 메서드
|
||||
# ════════════════════════════════════════════════════════════
|
||||
def get_token_row(self, mall_id: str) -> dict[str, Any] | None:
|
||||
with self._pool.connection() as conn:
|
||||
return conn.execute(
|
||||
"SELECT * FROM cafe24_oauth_tokens WHERE mall_id = %s",
|
||||
(mall_id,),
|
||||
).fetchone()
|
||||
|
||||
def save_token_row(self, *, mall_id: str, **fields: Any) -> None:
|
||||
with self._pool.connection() as conn:
|
||||
_update_token_row(conn, mall_id, fields)
|
||||
|
||||
@contextmanager
|
||||
def token_lock(self, mall_id: str) -> Iterator[TokenLock]:
|
||||
"""토큰 행을 FOR UPDATE 로 잠근 채 작업.
|
||||
|
||||
web 컨테이너와 worker 컨테이너가 동시에 refresh 하는 것을 막는다
|
||||
(카페24는 refresh token 을 회전시키므로 동시 갱신 시 한쪽이 무효화됨).
|
||||
행이 아직 없으면 row=None 으로 넘어간다.
|
||||
"""
|
||||
with self._pool.connection() as conn:
|
||||
with conn.transaction():
|
||||
row = conn.execute(
|
||||
"SELECT * FROM cafe24_oauth_tokens WHERE mall_id = %s FOR UPDATE",
|
||||
(mall_id,),
|
||||
).fetchone()
|
||||
yield TokenLock(conn, mall_id, row)
|
||||
|
||||
def disconnect(self, mall_id: str) -> None:
|
||||
"""연결 해제 — 토큰만 지운다(이력/예약은 보존)."""
|
||||
with self._pool.connection() as conn:
|
||||
conn.execute("DELETE FROM cafe24_oauth_tokens WHERE mall_id = %s", (mall_id,))
|
||||
|
||||
# ════════════════════════════════════════════════════════════
|
||||
# API 호출 로그 (Cafe24Client 가 주입받아 호출)
|
||||
# ⚠️ Authorization/토큰/시크릿은 절대 기록하지 않는다.
|
||||
# ════════════════════════════════════════════════════════════
|
||||
def log_api_call(
|
||||
self,
|
||||
*,
|
||||
endpoint: str,
|
||||
method: str,
|
||||
product_no: int | None,
|
||||
http_status: int | None,
|
||||
result: str,
|
||||
error_message: str,
|
||||
duration_ms: int,
|
||||
) -> None:
|
||||
with self._pool.connection() as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO cafe24_api_logs
|
||||
(endpoint, method, product_no, http_status, result, error_message, duration_ms)
|
||||
VALUES (%s,%s,%s,%s,%s,%s,%s)
|
||||
""",
|
||||
(endpoint, method, product_no, http_status, result, error_message, duration_ms),
|
||||
)
|
||||
|
||||
def list_api_logs(self, *, limit: int = 100) -> list[dict[str, Any]]:
|
||||
with self._pool.connection() as conn:
|
||||
rows = conn.execute(
|
||||
"""
|
||||
SELECT * FROM cafe24_api_logs
|
||||
ORDER BY created_at DESC, id DESC
|
||||
LIMIT %s
|
||||
""",
|
||||
(max(1, min(int(limit), 500)),),
|
||||
).fetchall()
|
||||
return [self._serialize(r) for r in rows]
|
||||
|
||||
# ════════════════════════════════════════════════════════════
|
||||
# 작업 감사 로그
|
||||
# ════════════════════════════════════════════════════════════
|
||||
def log_audit(
|
||||
self,
|
||||
*,
|
||||
actor: str,
|
||||
action: str,
|
||||
product_no: int | None = None,
|
||||
revision_id: int | None = None,
|
||||
schedule_id: int | None = None,
|
||||
result: str = "",
|
||||
detail: str = "",
|
||||
) -> None:
|
||||
with self._pool.connection() as conn:
|
||||
self._insert_audit(
|
||||
conn,
|
||||
actor=actor,
|
||||
action=action,
|
||||
product_no=product_no,
|
||||
revision_id=revision_id,
|
||||
schedule_id=schedule_id,
|
||||
result=result,
|
||||
detail=detail,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _insert_audit(
|
||||
conn: Any,
|
||||
*,
|
||||
actor: str,
|
||||
action: str,
|
||||
product_no: int | None = None,
|
||||
revision_id: int | None = None,
|
||||
schedule_id: int | None = None,
|
||||
result: str = "",
|
||||
detail: str = "",
|
||||
) -> None:
|
||||
"""호출자의 트랜잭션에 합류시키기 위해 conn 을 받는 정적 헬퍼."""
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO cafe24_audit_logs
|
||||
(actor, action, product_no, revision_id, schedule_id, result, detail)
|
||||
VALUES (%s,%s,%s,%s,%s,%s,%s)
|
||||
""",
|
||||
(actor, action, product_no, revision_id, schedule_id, result, detail[:1000]),
|
||||
)
|
||||
|
||||
def list_audit_logs(self, *, limit: int = 100) -> list[dict[str, Any]]:
|
||||
with self._pool.connection() as conn:
|
||||
rows = conn.execute(
|
||||
"""
|
||||
SELECT * FROM cafe24_audit_logs
|
||||
ORDER BY created_at DESC, id DESC
|
||||
LIMIT %s
|
||||
""",
|
||||
(max(1, min(int(limit), 500)),),
|
||||
).fetchall()
|
||||
return [self._serialize(r) for r in rows]
|
||||
|
||||
# ════════════════════════════════════════════════════════════
|
||||
# 직렬화 — datetime → KST ISO, date → ISO (다른 모듈과 동일)
|
||||
# ════════════════════════════════════════════════════════════
|
||||
@staticmethod
|
||||
def _serialize(row: dict[str, Any] | None) -> dict[str, Any]:
|
||||
if not row:
|
||||
return {}
|
||||
out = dict(row)
|
||||
for key, value in list(out.items()):
|
||||
if isinstance(value, datetime):
|
||||
aware = value if value.tzinfo else value.replace(tzinfo=KST)
|
||||
out[key] = aware.astimezone(KST).isoformat(timespec="seconds")
|
||||
elif isinstance(value, date):
|
||||
out[key] = value.isoformat()
|
||||
return out
|
||||
|
||||
|
||||
__all__ = ["Cafe24Store", "TokenLock", "store"]
|
||||
Reference in New Issue
Block a user