king
c6fb8ed375
카페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>
119 lines
3.9 KiB
Python
119 lines
3.9 KiB
Python
"""카페24 모듈 공용 가드/컨텍스트 헬퍼.
|
|
|
|
router.py 와 routes_*.py 가 함께 쓴다(순환 import 방지를 위해 분리).
|
|
다른 모듈과 동일한 규칙:
|
|
- JSON API → require_user() : 401/403 HTTPException
|
|
- HTML 페이지 → guard() : 리다이렉트 / denied.html 응답 반환
|
|
app.main 은 함수 안에서 지연 import 한다(순환 import 방지).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import HTTPException, Request
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
|
|
MODULE_KEY = "cafe24"
|
|
MODULE_NAME = "카페24 상품관리"
|
|
|
|
CONFIG_HELP = (
|
|
"카페24 모듈이 아직 설정되지 않았습니다. "
|
|
"CAFE24_DB_URL 환경변수를 설정하고 "
|
|
"scripts/sql/cafe24_db_init.sql 로 cafe24_db 를 초기화한 뒤 "
|
|
"컨테이너를 재기동하세요."
|
|
)
|
|
|
|
|
|
def get_store(request: Request) -> Any:
|
|
return getattr(request.app.state, "cafe24_store", None)
|
|
|
|
|
|
def require_user(request: Request) -> dict[str, Any]:
|
|
from app.main import get_current_user_record # noqa: WPS433
|
|
from app.store import has_module # noqa: WPS433
|
|
|
|
user = get_current_user_record(request)
|
|
if user is None:
|
|
raise HTTPException(status_code=401, detail="로그인이 필요합니다.")
|
|
if not has_module(user, MODULE_KEY):
|
|
raise HTTPException(status_code=403, detail=f"{MODULE_NAME} 모듈 권한이 없습니다.")
|
|
return user
|
|
|
|
|
|
def require_admin(request: Request) -> dict[str, Any]:
|
|
"""카페24 연결(OAuth)·연결 해제는 관리자만."""
|
|
from app.store import is_admin # noqa: WPS433
|
|
|
|
user = require_user(request)
|
|
if not is_admin(user):
|
|
raise HTTPException(status_code=403, detail="관리자만 카페24 연결을 변경할 수 있습니다.")
|
|
return user
|
|
|
|
|
|
def require_store(request: Request) -> tuple[Any, dict[str, Any]]:
|
|
"""JSON API 용 — store 미설정이면 503."""
|
|
user = require_user(request)
|
|
st = get_store(request)
|
|
if st is None:
|
|
raise HTTPException(status_code=503, detail=CONFIG_HELP)
|
|
return st, user
|
|
|
|
|
|
def render_config_needed(request: Request, user: dict[str, Any]) -> HTMLResponse:
|
|
from app.main import build_erp_nav, render_template # noqa: WPS433
|
|
from app.store import is_admin # noqa: WPS433
|
|
|
|
return render_template(
|
|
request,
|
|
"denied.html",
|
|
{
|
|
"reason": CONFIG_HELP,
|
|
"user": user,
|
|
"is_admin": is_admin(user),
|
|
"nav_items": build_erp_nav(user, active=MODULE_KEY),
|
|
},
|
|
status_code=503,
|
|
)
|
|
|
|
|
|
def guard(request: Request):
|
|
"""로그인+권한+store 점검. 페이지 핸들러 진입부에서 사용.
|
|
|
|
반환이 tuple 이면 (store, user), 아니면 그대로 응답으로 돌려준다.
|
|
"""
|
|
from app.main import get_current_user_record, render_template # noqa: WPS433
|
|
from app.store import has_module, is_admin # noqa: WPS433
|
|
|
|
user = get_current_user_record(request)
|
|
if user is None:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not has_module(user, MODULE_KEY):
|
|
return render_template(
|
|
request,
|
|
"denied.html",
|
|
{
|
|
"reason": f"{MODULE_NAME} 접근 권한이 없습니다.",
|
|
"user": user,
|
|
"is_admin": is_admin(user),
|
|
},
|
|
status_code=403,
|
|
)
|
|
st = get_store(request)
|
|
if st is None:
|
|
return render_config_needed(request, user)
|
|
return st, user
|
|
|
|
|
|
def base_ctx(request: Request, user: dict[str, Any], *, active_tab: str = "") -> dict[str, Any]:
|
|
from app.main import build_erp_nav # noqa: WPS433
|
|
from app.store import is_admin # noqa: WPS433
|
|
|
|
return {
|
|
"user": user,
|
|
"is_admin": is_admin(user),
|
|
"is_super": bool(user.get("is_super_admin")),
|
|
"nav_items": build_erp_nav(user, active=MODULE_KEY),
|
|
"active_tab": active_tab,
|
|
}
|