From bebb6a71e6771773fff55bb288a8a21a1aa47216 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EB=AF=BC=ED=97=8C?= Date: Wed, 8 Jul 2026 16:49:28 +0900 Subject: [PATCH] =?UTF-8?q?feat(lps):=20=EA=B2=80=EC=83=89=20API=20?= =?UTF-8?q?=EB=9D=BC=EC=9A=B0=ED=84=B0=20=E2=80=94=20=EC=9A=94=EC=B2=AD=20?= =?UTF-8?q?=EC=A0=81=EC=9E=AC(enqueue)=20+=20=EC=83=81=ED=83=9C/=ED=81=90?= =?UTF-8?q?=20=ED=86=B5=EA=B3=84=20=EC=A1=B0=ED=9A=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 커머스 검색요청 계약을 프레임워크(protocol+service+RemoveNoneResponse)로 재구성. 실제 검색은 워커가 큐에서 꺼내 수행하도록 API 는 적재까지만 담당(비동기 분리). - POST /v1/lps/search: 상품 리스트 → 상품별 SEARCH 잡 적재, product_code 로 활성 중복 방지, job_type→우선순위 매핑 - GET /v1/lps/jobs/{job_id}: 잡 상태/시도/결과 조회 - GET /v1/lps/queue/stats: 상태별 카운트(모니터링) - protocol/lps_service 추가, job_crud.get() 단건조회, enums LPS_JOB_NOT_FOUND - tests: 적재/중복/상태/미존재/통계 5건 (ASGI 클라이언트 + 실 lps_db) → 전체 16/16 Co-Authored-By: Claude Opus 4.8 (1M context) --- lps/common/enums.py | 3 ++ lps/crud/job_crud.py | 21 +++++++++++ lps/router/router.py | 7 ++-- lps/router/v1/lps/protocol.py | 47 ++++++++++++++++++++++++ lps/router/v1/lps/search.py | 39 ++++++++++++++++++++ lps/services/lps_service.py | 67 +++++++++++++++++++++++++++++++++++ lps/tests/test_lps_api.py | 61 +++++++++++++++++++++++++++++++ 7 files changed, 240 insertions(+), 5 deletions(-) create mode 100644 lps/router/v1/lps/protocol.py create mode 100644 lps/router/v1/lps/search.py create mode 100644 lps/services/lps_service.py create mode 100644 lps/tests/test_lps_api.py diff --git a/lps/common/enums.py b/lps/common/enums.py index f601b66..b275871 100644 --- a/lps/common/enums.py +++ b/lps/common/enums.py @@ -29,6 +29,9 @@ class ErrorType(Enum): HTTP_TO_MANY_REQUEST = 429 HTTP_INVALID_CLIENT_ACCESS = 433 + # LPS 도메인 에러 (1500~) + LPS_JOB_NOT_FOUND = 1500 # 잡 없음/잘못된 job_id + # ErrorType 의 HTTP_* 값과 status_code 를 맞춰 router 단에서 raise 한다. EXCEPTION_INVALID_CLIENT_REQUEST = HTTPException(status_code=ErrorType.HTTP_INVALID_CLIENT_REQUEST.value, detail=ErrorType.HTTP_INVALID_CLIENT_REQUEST.name) diff --git a/lps/crud/job_crud.py b/lps/crud/job_crud.py index 9404e50..dcd4001 100644 --- a/lps/crud/job_crud.py +++ b/lps/crud/job_crud.py @@ -163,6 +163,27 @@ class JobQueue: return await self._tx(run) + # ---- 단건 조회 ------------------------------------------------------ + async def get(self, job_id: str) -> dict | None: + """잡 단건 조회(읽기). 없으면 None. status 는 정수(JobStatus 값).""" + sql = text(""" + SELECT job_id, job_type, status, priority, attempts, max_attempts, + result, last_error, run_after, created_at, updated_at + FROM job WHERE job_id = :id + """) + s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value) + try: + row = (await s.execute(sql, {"id": job_id})).mappings().first() + if not row: + return None + d = dict(row) + d["job_id"] = str(d["job_id"]) + if isinstance(d.get("result"), str): + d["result"] = json.loads(d["result"]) + return d + finally: + await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value) + # ---- 관측(관리 API/메트릭용) --------------------------------------- async def counts(self) -> dict[str, int]: """상태별 잡 개수(관리 API·알림용). 수동 psql 스크립트를 대체한다.""" diff --git a/lps/router/router.py b/lps/router/router.py index cb073b6..0db0df1 100644 --- a/lps/router/router.py +++ b/lps/router/router.py @@ -9,10 +9,7 @@ from common.database.db_session_manager import DB_SESSION_MNG from common.logger import LOG from common.utils.gtime import GTime from config.server_configs import web_server_config - -# 도메인 라우터가 생기면 아래처럼 import 후 include 한다(backend 컨벤션): -# import router.v1.. -# app.include_router(router.v1...router) +import router.v1.lps.search API_SERVER_START_TIME = GTime.UTCStr() @@ -62,4 +59,4 @@ async def healthz(): # 각 도메인 라우터를 등록한다. 새 기능 추가 시 router.v1.. 를 import 후 include. -# (아직 도메인 로직 미정 — healthz 만 노출) +app.include_router(router.v1.lps.search.router) diff --git a/lps/router/v1/lps/protocol.py b/lps/router/v1/lps/protocol.py new file mode 100644 index 0000000..cfa0c28 --- /dev/null +++ b/lps/router/v1/lps/protocol.py @@ -0,0 +1,47 @@ +"""LPS API 요청/응답 프로토콜. 커머스→오투오 검색요청 계약을 미러링한다.""" + +from typing import Optional + +from pydantic import BaseModel, Field + +from common.models.gmodel import Req_WebPacketProtocol, Res_WebPacketProtocol + + +class SearchItem(BaseModel): + """검색 대상 상품 1건. 규격/모델/제조사는 매칭(향후 AI 유사도) 입력으로 함께 적재한다.""" + + product_code: str = Field(description="상품 식별 코드(커머스 기준). dedupe 키로도 사용") + product_name: str = Field(description="상품명") + job_type: str = Field("single", description="요청 유형: new|single|negowiz|batch → 우선순위 매핑") + model: str = Field("", description="모델명") + specification: str = Field("", description="규격(용량/개입/수량 등)") + company: str = Field("", description="제조사/브랜드") + price: str = Field("", description="현재가(참고, 문자열)") + + +class Req_Search(Req_WebPacketProtocol): + data: list[SearchItem] = Field(description="검색 대상 상품 리스트") + + +class EnqueuedItem(BaseModel): + product_code: str + job_id: Optional[str] = Field(None, description="적재된 잡 ID. 활성 중복이면 None") + duplicated: bool = Field(False, description="활성 중복(PENDING/RUNNING)이라 스킵됐는지") + + +class Res_Search(Res_WebPacketProtocol): + accepted: int = Field(0, description="새로 적재된 잡 수(중복 제외)") + items: list[EnqueuedItem] = Field(default_factory=list) + + +class Res_JobStatus(Res_WebPacketProtocol): + job_id: Optional[str] = None + status: Optional[str] = Field(None, description="JobStatus 이름(PENDING/RUNNING/DONE/DEAD)") + attempts: Optional[int] = None + max_attempts: Optional[int] = None + output: Optional[dict] = Field(None, description="잡 결과(완료 시). 봉투 result 와 구분") + last_error: Optional[str] = None + + +class Res_QueueStats(Res_WebPacketProtocol): + counts: dict[str, int] = Field(default_factory=dict, description="상태별 잡 개수") diff --git a/lps/router/v1/lps/search.py b/lps/router/v1/lps/search.py new file mode 100644 index 0000000..22dc1c1 --- /dev/null +++ b/lps/router/v1/lps/search.py @@ -0,0 +1,39 @@ +"""LPS 검색 API 라우터. 요청 적재(enqueue) + 잡 상태/큐 통계 조회.""" + +from fastapi import APIRouter, Depends + +from router.v1.validator.dependencies import RemoveNoneResponse +from services.lps_service import LpsService +from router.v1.lps.protocol import Res_JobStatus, Res_QueueStats, Res_Search, Req_Search + +router = APIRouter(prefix="/v1/lps", tags=["LPS"], responses={404: {"description": "Not found"}}) + + +@router.post( + path="/search", + response_model=Res_Search, + summary="최저가 검색 요청", + description="상품 리스트를 받아 상품별 검색 잡을 큐에 적재한다(product_code 로 활성 중복 방지). 실제 검색은 워커가 비동기 수행.", +) +async def search(req: Req_Search, service: LpsService = Depends()): + return RemoveNoneResponse(await service.submit_search(req.data)) + + +@router.get( + path="/jobs/{job_id}", + response_model=Res_JobStatus, + summary="잡 상태 조회", + description="적재된 검색 잡의 상태/시도횟수/결과를 조회한다.", +) +async def job_status(job_id: str, service: LpsService = Depends()): + return RemoveNoneResponse(await service.get_job(job_id)) + + +@router.get( + path="/queue/stats", + response_model=Res_QueueStats, + summary="큐 상태 카운트", + description="상태별(PENDING/RUNNING/DONE/DEAD) 잡 개수. 관리/모니터링용.", +) +async def queue_stats(service: LpsService = Depends()): + return RemoveNoneResponse(await service.stats()) diff --git a/lps/services/lps_service.py b/lps/services/lps_service.py new file mode 100644 index 0000000..7620f66 --- /dev/null +++ b/lps/services/lps_service.py @@ -0,0 +1,67 @@ +"""LPS 검색 요청 도메인 로직. 요청을 검색 잡으로 큐에 적재하고 상태/통계를 조회한다. + +라우터는 요청 검증→service 호출→RemoveNoneResponse 만 담당(backend 컨벤션). +실제 검색/크롤링은 워커가 큐에서 잡을 꺼내 파이프라인으로 수행한다(비동기 분리). +""" + +import uuid + +from fastapi import Depends + +from common.enums import ErrorType, JobStatus, JobType +from crud.job_crud import JobQueue +from router.v1.lps.protocol import ( + EnqueuedItem, + Res_JobStatus, + Res_QueueStats, + Res_Search, + SearchItem, +) + +# 요청 유형 → 우선순위(낮을수록 우선). 알 수 없는 유형은 가장 낮은 우선순위(배치와 동급). +_PRIORITY = {"new": 1, "single": 2, "negowiz": 3, "batch": 4} + + +class LpsService: + def __init__(self, queue: JobQueue = Depends(JobQueue)): + self.queue = queue + + async def submit_search(self, items: list[SearchItem]) -> Res_Search: + res = Res_Search() + for it in items: + priority = _PRIORITY.get(it.job_type, 4) + job_id = await self.queue.enqueue( + JobType.SEARCH.value, + it.model_dump(), + priority=priority, + dedupe_key=f"search-{it.product_code}", + ) + res.items.append(EnqueuedItem(product_code=it.product_code, job_id=job_id, duplicated=job_id is None)) + res.accepted = sum(1 for i in res.items if i.job_id is not None) + return res + + async def get_job(self, job_id_str: str) -> Res_JobStatus: + res = Res_JobStatus() + try: + uuid.UUID(job_id_str) # 잘못된 id 는 DB 조회 전에 컷 + except (ValueError, TypeError): + res.result.SetResult(ErrorType.LPS_JOB_NOT_FOUND) + return res + + row = await self.queue.get(job_id_str) + if row is None: + res.result.SetResult(ErrorType.LPS_JOB_NOT_FOUND) + return res + + res.job_id = row["job_id"] + res.status = JobStatus(row["status"]).name + res.attempts = row["attempts"] + res.max_attempts = row["max_attempts"] + res.output = row.get("result") + res.last_error = row.get("last_error") + return res + + async def stats(self) -> Res_QueueStats: + res = Res_QueueStats() + res.counts = await self.queue.counts() + return res diff --git a/lps/tests/test_lps_api.py b/lps/tests/test_lps_api.py new file mode 100644 index 0000000..0b52cc1 --- /dev/null +++ b/lps/tests/test_lps_api.py @@ -0,0 +1,61 @@ +"""LPS API 라우터 테스트 — 검색 적재/중복/상태조회/큐 통계 (ASGI 클라이언트 + 실 lps_db).""" + +import pytest_asyncio +from sqlalchemy import text + + +@pytest_asyncio.fixture +async def clean_jobs(db_engine): + async with db_engine.begin() as conn: + await conn.execute(text("TRUNCATE job")) + + +async def test_search_enqueues_jobs(client, clean_jobs): + body = {"data": [ + {"product_code": "P1", "product_name": "커피", "job_type": "new"}, + {"product_code": "P2", "product_name": "무선마우스", "specification": "M170"}, + ]} + r = await client.post("/v1/lps/search", json=body) + assert r.status_code == 200 + j = r.json() + assert j["accepted"] == 2 + assert {i["product_code"] for i in j["items"]} == {"P1", "P2"} + assert all(i.get("job_id") for i in j["items"]) + + +async def test_search_dedupes_active_product(client, clean_jobs): + body = {"data": [{"product_code": "P1", "product_name": "커피", "job_type": "new"}]} + await client.post("/v1/lps/search", json=body) + r2 = await client.post("/v1/lps/search", json=body) # 같은 product_code 재요청 + item = r2.json()["items"][0] + assert item["duplicated"] is True + assert "job_id" not in item # None → RemoveNoneResponse 로 제거됨 + assert r2.json()["accepted"] == 0 + + +async def test_job_status_flow(client, clean_jobs): + jid = (await client.post("/v1/lps/search", json={"data": [{"product_code": "P9", "product_name": "커피"}]})).json()["items"][0]["job_id"] + r = await client.get(f"/v1/lps/jobs/{jid}") + body = r.json() + assert body["status"] == "PENDING" and body["attempts"] == 0 + assert body["result"]["success"] is True + + +async def test_job_status_not_found(client, clean_jobs): + # 존재하지 않는(유효 UUID) 잡 + r = await client.get("/v1/lps/jobs/00000000-0000-0000-0000-000000000000") + assert r.json()["result"]["success"] is False + assert r.json()["result"]["desc"] == "LPS_JOB_NOT_FOUND" + # 잘못된 형식의 id 도 not-found 처리 + r2 = await client.get("/v1/lps/jobs/not-a-uuid") + assert r2.json()["result"]["desc"] == "LPS_JOB_NOT_FOUND" + + +async def test_queue_stats(client, clean_jobs): + await client.post("/v1/lps/search", json={"data": [ + {"product_code": "A", "product_name": "x"}, + {"product_code": "B", "product_name": "y"}, + ]}) + r = await client.get("/v1/lps/queue/stats") + counts = r.json()["counts"] + assert counts["PENDING"] == 2 and counts["DONE"] == 0 and counts["DEAD"] == 0