feat(lps): 인터넷 최저가 검색 솔루션 프레임워크 골격 추가

backend 와 동일한 프레임워크로 lps/ 폴더를 신설한다. 구체적인 도메인
로직(크롤링/오픈API/최저가 산정)은 미정이라 골격만 구성한다.

- FastAPI 부트스트랩(web_main/router) + lifespan·CORS·gzip·로그 미들웨어 + /healthz
- TOML 설정 로더(APP_ENV) + pydantic config + DB env override
- DB 세션 매니저(논리 DB × R/W, service→람다 위임) — 엔진 lazy 라 DB 없이도 부팅
- 공통 응답 규약(gmodel) + ErrorType/DBType/DBWRType
- router→services→crud 계층 컨벤션(services/crud 는 빈 폴더로 자리만)
- 포트 9400(backend 9300·agent 9500 회피), Dockerfile/run_local_server.sh/README
- tests/test_health.py 스모크(healthz, DB 없이 통과)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

lps/.dockerignore

@@ -0,0 +1,9 @@
+.venv/
+venv/
+__pycache__/
+*.pyc
+.pytest_cache/
+.git/
+tests/
+loadtest/
+*.md

lps/Dockerfile

@@ -0,0 +1,16 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+# 의존성 먼저 설치 (레이어 캐시 활용)
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+# 항상 APP_ENV=local 로 실행 → config.local.toml 사용.
+ENV APP_ENV=local
+
+EXPOSE 9400
+
+CMD ["python", "web_main.py"]

lps/README.md

@@ -0,0 +1,56 @@
+# LPS (Lowest Price Search)
+
+인터넷 최저가를 찾는 솔루션. **프레임워크는 `backend` 와 동일**하게 구성한 골격이며,
+구체적인 도메인 로직(크롤링/오픈API 연동/최저가 산정 등)은 아직 정해지지 않았다.
+
+## 스택 / 아키텍처 (backend 미러링)
+- **FastAPI** 앱 (`router/router.py`) + `web_main.py` 부트스트랩
+- **설정**: `config/` — TOML 로더(`config.local.toml`) + pydantic 모델, `APP_ENV`(기본 local)
+- **DB**: `common/database/db_session_manager.py` — 논리 DB × Read/Write 엔진, service→람다 위임 패턴
+  - 엔진은 lazy 생성이라 **DB 없이도 부팅/healthz 동작**한다. 테이블·crud 가 생기면 그때 실제 접속.
+- **공통 응답 규약**: `common/models/gmodel.py` — `Res_WebPacketProtocol.result`(성공/코드/설명)
+- **결과 코드**: `common/enums.py` — `ErrorType`, `DBType`, `DBWRType`
+- 계층 컨벤션: `router`(컨트롤러) → `services`(비즈니스) → `crud`(DB 접근)
+
+## 폴더 구조
+```
+lps/
+├── web_main.py                     # 진입점
+├── requirements.txt / Dockerfile / .dockerignore
+├── run_local_server.sh             # 로컬 실행(대화형), 포트 9400
+├── pytest.ini / conftest.py
+├── config/
+│   ├── config_loader.py / config_models.py / server_configs.py
+│   └── config.local.toml.example   # cp 해서 config.local.toml 로 사용(시크릿, 미커밋)
+├── common/
+│   ├── enums.py / logger.py / singleton.py
+│   ├── utils/gtime.py
+│   ├── models/gmodel.py            # 프로토콜 base
+│   └── database/{db_session_manager.py, model/models.py(MAIN_BASE)}
+├── router/
+│   ├── router.py                   # app + /healthz (도메인 라우터 미등록)
+│   └── v1/validator/dependencies.py# RemoveNoneResponse
+├── services/                       # (비어있음) 도메인 서비스 추가 위치
+├── crud/                           # (비어있음) DB 접근 계층 추가 위치
+└── tests/test_health.py            # 스모크 테스트
+```
+
+## 로컬 실행
+```bash
+cp config/config.local.toml.example config/config.local.toml   # 값 채우기
+./run_local_server.sh                                          # → http://localhost:9400/docs
+```
+
+## 테스트
+```bash
+python -m pytest            # tests/ (기본 healthz 스모크)
+```
+
+## 포트
+- backend 9300 / agent 9500 과 겹치지 않도록 **LPS 는 9400** 사용.
+
+## 새 도메인 추가 순서 (backend 컨벤션)
+1. `common/database/model/models.py` 에 테이블 정의(+ `DBType`)
+2. `crud/<domain>_crud.py` (ABC 인터페이스 + 구현)
+3. `services/<domain>_service.py` (비즈니스 로직)
+4. `router/v1/<domain>/{protocol.py, <domain>.py}` 작성 후 `router/router.py` 에서 `include_router`

lps/common/database/db_session_manager.py

@@ -0,0 +1,206 @@
+from asyncio import current_task
+
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_scoped_session
+from sqlalchemy.util._collections import immutabledict
+
+from common.database.model.models import MAIN_BASE
+from common.enums import DBType, DBWRType, ErrorType
+from common.logger import LOG
+from common.singleton import Singleton
+from config.server_configs import main_db_config
+
+
+class DBSessionManager(Singleton):
+    """DB 세션/엔진 관리자 (싱글톤).
+
+    핵심 패턴
+    - DBType(논리 DB) x DBWRType(Read/Write) 조합마다 별도 async 엔진을 둔다.
+      => 조회는 Read 복제본, 변경은 Write 주 DB 로 자연스럽게 분리된다.
+    - 비즈니스 로직(service)은 직접 세션을 열지 않고 "람다"를 넘긴다.
+      execute_lambda        : 단일 쿼리 (주로 조회)
+      execute_lambda_run    : 동일 DB 의 여러 변경 쿼리를 한 트랜잭션으로 commit
+      세션 open/close 와 commit/rollback 은 매니저가 책임진다.
+    - 엔진 생성은 lazy 하다(create_async_engine 은 실제 커넥션을 맺지 않음). DB 없이도 import/부팅 가능.
+    """
+
+    def __init__(self):
+        if DBSessionManager.is_init():
+            LOG.e_no_callstack("already init DBSessionManager")
+            return
+        DBSessionManager.set_init()
+
+        self.__DB_URL_MAP = {"postgresql": "postgresql+asyncpg"}
+        # 종료 시 dispose 하기 위해 생성한 엔진을 모아둔다.
+        self.__engines = []
+        # 논리 DB -> config. DB 가 늘어나면 여기에 추가만 하면 된다.
+        self.__db_type_map = {
+            DBType.MAIN.value: main_db_config,
+        }
+
+        # Write 엔진 맵
+        self.__write_session = {
+            DBType.MAIN.value: self.create_engine(DBType.MAIN.value, DBWRType.DB_WRITE.value),
+        }
+        # Read 엔진 맵
+        self.__read_session = {
+            DBType.MAIN.value: self.create_engine(DBType.MAIN.value, DBWRType.DB_READ.value),
+        }
+
+    def create_engine(self, db_type: int, db_wr_type: int):
+        db_config = self.__db_type_map.get(db_type)
+        if not db_config:
+            raise ValueError("Invalid database type")
+
+        if db_wr_type == DBWRType.DB_READ.value:
+            pw = (":" + db_config.read_pw) if len(db_config.read_pw) > 0 else ""
+            db_url = f"{self.__DB_URL_MAP[db_config.db_type]}://{db_config.read_id}{pw}@{db_config.read_host}:{db_config.read_port}/{db_config.name}"
+            LOG.i(f"Read DB create engine url : {db_url}")
+        else:
+            pw = (":" + db_config.write_pw) if len(db_config.write_pw) > 0 else ""
+            db_url = f"{self.__DB_URL_MAP[db_config.db_type]}://{db_config.write_id}{pw}@{db_config.write_host}:{db_config.write_port}/{db_config.name}"
+            LOG.i(f"Write DB create engine url : {db_url}")
+
+        # SSL/TLS: 관리형 DB(RDS/Aurora/Azure)는 보통 TLS 필수. sslmode 가 설정되면 asyncpg 에 전달.
+        connect_args = {}
+        sslmode = (getattr(db_config, "sslmode", "") or "").lower()
+        if sslmode and sslmode != "disable":
+            connect_args["ssl"] = sslmode
+
+        engine = create_async_engine(
+            db_url,
+            echo=db_config.show_log,
+            pool_size=db_config.pool_size,
+            max_overflow=db_config.max_overflow,
+            pool_pre_ping=True,
+            pool_recycle=600,
+            connect_args=connect_args,
+        )
+        self.__engines.append(engine)
+        scoped_session = async_scoped_session(
+            sessionmaker(engine, class_=AsyncSession, expire_on_commit=False, autocommit=False, autoflush=False),
+            scopefunc=current_task,
+        )
+        return scoped_session
+
+    async def dispose_all(self):
+        """모든 엔진의 커넥션 풀을 정리한다. 앱 종료/테스트 종료 시 호출한다.
+        호출하지 않으면 풀 커넥션이 이벤트 루프 종료 후 GC 되며 경고를 남긴다.
+        """
+        for engine in self.__engines:
+            await engine.dispose()
+
+    # ---- 세션 lifecycle -------------------------------------------------
+    async def start_session(self, db_type: int, db_wr_type: int) -> AsyncSession:
+        if db_wr_type == DBWRType.DB_WRITE.value:
+            return self.__write_session[db_type]()
+        return self.__read_session[db_type]()
+
+    async def end_session(self, db_type: int, db_wr_type: int):
+        if db_wr_type == DBWRType.DB_WRITE.value:
+            await self.__write_session[db_type].remove()
+        else:
+            await self.__read_session[db_type].remove()
+
+    # ---- 저수준 DB 연산 (crud 에서 호출) --------------------------------
+    async def run(self, db: AsyncSession, err_msg="DB Run Failed", raise_error=True) -> ErrorType:
+        try:
+            await db.commit()
+            return ErrorType.SUCCESS
+        except IntegrityError as ex:
+            await db.rollback()
+            LOG.e_no_callstack(f"duplicated. {ex}")
+            return ErrorType.DB_ALREADY_SAME_KEY
+        except Exception as ex:
+            await db.rollback()
+            err_type = ErrorType.DB_RUN_FAILED
+            LOG.e_no_callstack(f"[{err_type.name}] {err_msg=}, {ex=}")
+            if raise_error:
+                raise RuntimeError(err_type.name, err_msg)
+            return err_type
+
+    async def insert(self, db: AsyncSession, obj, err_msg="DB Failed", raise_error=True) -> ErrorType:
+        try:
+            if isinstance(obj, MAIN_BASE):
+                db.add(obj)
+            elif isinstance(obj, list):
+                db.add_all(obj)
+            else:
+                raise RuntimeError("DO NOT USE QUERY IN DBJOB")
+            return ErrorType.SUCCESS
+        except Exception as ex:
+            await db.rollback()
+            err_type = ErrorType.DB_RUN_FAILED
+            LOG.e_no_callstack(f"[{err_type.name}] {err_msg=}, {ex=}")
+            if raise_error:
+                raise RuntimeError(err_type.name, err_msg)
+            return err_type
+
+    async def add(self, db: AsyncSession, query, err_msg="DB Operation Failed", raise_error=True) -> ErrorType:
+        """update/delete 등 비-select 쿼리 실행."""
+        try:
+            if hasattr(query, "column_descriptions"):
+                raise RuntimeError("DO NOT USE SELECT QUERY IN DBJOB")
+            await db.execute(query, execution_options=immutabledict({"synchronize_session": "fetch"}))
+            return ErrorType.SUCCESS
+        except IntegrityError as ex:
+            await db.rollback()
+            err_type = ErrorType.DB_ALREADY_SAME_KEY
+            LOG.e_no_callstack(f"[{err_type.name}] {err_msg=}, {ex=}")
+            return err_type
+        except Exception as ex:
+            await db.rollback()
+            err_type = ErrorType.DB_RUN_FAILED
+            LOG.e_no_callstack(f"[{err_type.name}] {err_msg=}, {ex=}")
+            if raise_error:
+                raise RuntimeError(err_type.name, err_msg)
+            return err_type
+
+    async def execute(self, db: AsyncSession, query, err_msg="DB Query Execution Failed", raise_error=True) -> tuple[ErrorType, list]:
+        """select 쿼리 실행 후 결과 리스트 반환."""
+        try:
+            if not hasattr(query, "column_descriptions"):
+                raise RuntimeError("DO NOT USE NON-SELECT QUERY IN DBJOB")
+            res = await db.execute(query, execution_options=immutabledict({"synchronize_session": "fetch"}))
+            return ErrorType.SUCCESS, res.scalars().fetchall() if 1 == len(query.column_descriptions) else res.all()
+        except Exception as ex:
+            err_type = ErrorType.DB_RUN_FAILED
+            LOG.e_no_callstack(f"[{err_type.name}] {err_msg=}, {ex=}")
+            if raise_error:
+                raise RuntimeError(err_type.name, err_msg)
+            return err_type, []
+
+    # ---- 람다 실행 진입점 (service 에서 호출) ---------------------------
+    async def execute_lambda(self, db_type: int, db_wr_type: int, func):
+        """단일 쿼리 호출. func(session) 한 개를 실행하고 결과를 그대로 반환."""
+        s = await self.start_session(db_type, db_wr_type)
+        try:
+            return await func(s)
+        finally:
+            await self.end_session(db_type, db_wr_type)
+
+    async def execute_lambda_run(self, db_type_list: list[int], func_list: list):
+        """동일 DB 의 변경 쿼리 여러 개를 한 트랜잭션으로 실행 후 commit.
+        하나라도 SUCCESS 가 아니면 즉시 중단(rollback)된다.
+        """
+        temp_list = list(set(db_type_list))
+        if len(temp_list) != 1:
+            return ErrorType.DB_INVALID_TYPE
+
+        db_type = temp_list[0]
+        s = await self.start_session(db_type, DBWRType.DB_WRITE.value)
+        try:
+            for func in func_list:
+                err_type = await func(s)
+                if err_type != ErrorType.SUCCESS:
+                    return err_type
+            return await self.run(s)
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED
+        finally:
+            await self.end_session(db_type, DBWRType.DB_WRITE.value)
+
+
+DB_SESSION_MNG = DBSessionManager()

lps/common/database/model/models.py

@@ -0,0 +1,8 @@
+from sqlalchemy.orm import declarative_base
+
+# 모든 ORM 모델의 베이스. insert 시 isinstance 체크에도 사용된다.
+# 도메인 테이블이 생기면 아래에 backend/common/database/model/models.py 스타일로 정의한다.
+#   - @staticmethod DBType() 로 소속 논리 DB(common.enums.DBType)를 반환
+#   - 코드값(status/type 등)은 SMALLINT 정수 코드(앱 enum 매핑)
+#   - 소프트 삭제(deleted), created_at/updated_at 컨벤션 유지
+MAIN_BASE = declarative_base()

lps/common/enums.py

@@ -0,0 +1,51 @@
+from enum import Enum, auto
+
+from fastapi import HTTPException
+
+
+class ErrorType(Enum):
+    """서버 전역 결과 코드. Res_WebPacketProtocol.result 에 담겨 클라이언트로 전달된다.
+    HTTP status 와 겹치지 않도록 구간을 분리해서 관리한다.
+    도메인 로직이 생기면 각 구간(예: 1500~ LPS 전용)을 이어서 추가한다.
+    """
+
+    SUCCESS = 0
+    FAIL = 1
+
+    # DB 에러
+    DB_RUN_FAILED = 10
+    DB_ALREADY_SAME_KEY = auto()
+    DB_INVALID_KEY = auto()
+    DB_EMPTY_DATA = auto()
+    DB_INVALID_TYPE = auto()
+
+    # 요청/직렬화 에러
+    JSON_PARSE_ERROR = 100
+    INVALID_REQUEST_DATA = auto()
+    INTERNAL_EXCEPTION = auto()
+
+    # http 에러 코드와 겹치지 않게 설정 - router 전용 예외 발생 옵션
+    HTTP_INVALID_CLIENT_REQUEST = 419
+    HTTP_TO_MANY_REQUEST = 429
+    HTTP_INVALID_CLIENT_ACCESS = 433
+
+
+# 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)
+EXCEPTION_TO_MANY_REQUEST = HTTPException(status_code=ErrorType.HTTP_TO_MANY_REQUEST.value, detail=ErrorType.HTTP_TO_MANY_REQUEST.name)
+EXCEPTION_INVALID_CLIENT_ACCESS = HTTPException(status_code=ErrorType.HTTP_INVALID_CLIENT_ACCESS.value, detail=ErrorType.HTTP_INVALID_CLIENT_ACCESS.name)
+
+
+class DBType(Enum):
+    """논리 DB 식별자. 물리적으로 같은 DB 라도 도메인별로 논리 구분한다.
+    DB 가 늘어나면 여기에 추가하고 db_session_manager 의 엔진 맵에도 등록한다.
+    """
+
+    MAIN = 1  # LPS 기본 DB
+
+
+class DBWRType(Enum):
+    """Read/Write 분리. 조회는 READ(복제), 변경은 WRITE(주 DB)."""
+
+    DB_READ = 1
+    DB_WRITE = 2

lps/common/logger.py

@@ -0,0 +1,43 @@
+import sys
+import traceback
+from datetime import datetime, timezone
+
+
+class _Logger:
+    """원본 DerbyServer LOG 인터페이스를 간소화한 버전.
+    LOG.i / LOG.d / LOG.w / LOG.e_no_callstack / LOG.SetPrefix 를 제공한다.
+    """
+
+    def __init__(self):
+        self._prefix = ""
+
+    def SetPrefix(self, prefix: str):
+        self._prefix = prefix
+
+    def _now(self) -> str:
+        return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
+
+    def _write(self, level: str, msg):
+        head = f"{self._now()} [{level}]"
+        if self._prefix:
+            head += f"[{self._prefix}]"
+        print(f"{head} {msg}", file=sys.stderr if level in ("WARN", "ERROR") else sys.stdout)
+
+    def d(self, msg):
+        self._write("DEBUG", msg)
+
+    def i(self, msg):
+        self._write("INFO", msg)
+
+    def w(self, msg):
+        self._write("WARN", msg)
+
+    def e(self, msg):
+        self._write("ERROR", msg)
+        traceback.print_stack()
+
+    def e_no_callstack(self, msg):
+        self._write("ERROR", msg)
+
+
+LOG = _Logger()

lps/common/models/gmodel.py

@@ -0,0 +1,44 @@
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+from common.enums import ErrorType
+
+
+class StructModel:
+    """프로토콜/구조체 식별용 마커 클래스."""
+
+    pass
+
+
+class ErrorInfo(BaseModel, StructModel):
+    """모든 응답에 공통으로 실리는 결과 정보. result.success / code / desc 로 내려간다."""
+
+    success: Optional[bool] = Field(True, description="처리 성공 여부 (성공 시 true)")
+    code: Optional[int] = Field(ErrorType.SUCCESS.value, description="결과 코드 (ErrorType, 0=성공)")
+    desc: Optional[str] = Field(ErrorType.SUCCESS.name, description="결과 코드 이름 (ErrorType.name)")
+
+    def SetResult(self, enum: ErrorType):
+        if enum is not None:
+            self.success = ErrorType.SUCCESS.value == enum.value
+            self.code = enum.value
+            self.desc = enum.name
+
+
+# ---- Protocol 규약 -------------------------------------------------------
+# 모든 통신 패킷은 WebPacketProtocol 을 상속한다.
+# 요청   : Req_xxx (Req_WebPacketProtocol)
+# 응답   : Res_xxx (Res_WebPacketProtocol) - 항상 result 필드를 가진다.
+# 각 라우터 폴더의 protocol.py 에 Req_/Res_ 를 정의한다.
+class WebPacketProtocol(BaseModel, StructModel):
+    pass
+
+
+class Req_WebPacketProtocol(WebPacketProtocol):
+    pass
+
+
+class Res_WebPacketProtocol(WebPacketProtocol):
+    # default_factory 로 인스턴스마다 새 ErrorInfo 를 생성한다 (mutable default 공유 방지).
+    result: ErrorInfo = Field(default_factory=ErrorInfo, description="공통 처리 결과 (성공 여부/코드/설명)")
+    msg: Optional[str] = Field(None, description="부가 메시지 (선택)")

lps/common/singleton.py

@@ -0,0 +1,16 @@
+class Singleton:
+    _init = False
+
+    def __new__(cls, *args, **kwargs):
+        if not hasattr(cls, "instance"):
+            cls.instance = super(Singleton, cls).__new__(cls)
+
+        return cls.instance
+
+    @classmethod
+    def is_init(cls):
+        return cls._init
+
+    @classmethod
+    def set_init(cls):
+        cls._init = True

lps/common/utils/gtime.py

@@ -0,0 +1,21 @@
+from datetime import datetime, timezone, timedelta
+
+
+class GTime:
+    """서버 전역에서 UTC 기준 시간을 사용하기 위한 유틸. (원본 DerbyServer 패턴 축약)"""
+
+    @staticmethod
+    def UTC() -> datetime:
+        return datetime.now(timezone.utc).replace(tzinfo=None)
+
+    @staticmethod
+    def UTCStr(fmt: str = "%Y-%m-%d %H:%M:%S") -> str:
+        return GTime.UTC().strftime(fmt)
+
+    @staticmethod
+    def AddMinutes(minutes: int) -> datetime:
+        return GTime.UTC() + timedelta(minutes=minutes)
+
+    @staticmethod
+    def AddDays(days: int) -> datetime:
+        return GTime.UTC() + timedelta(days=days)

lps/config/config.local.toml.example

@@ -0,0 +1,34 @@
+# 복사해서 사용: cp config.local.toml.example config.local.toml
+# 실제 config.local.toml 은 시크릿 포함이라 커밋하지 않는다(.gitignore: *.toml).
+# 모든 서버는 APP_ENV=local 로 띄우며 이 파일을 읽는다.
+[WebServerConfig]
+server_name = "LpsServer"
+port = 9400
+process_count = 1
+is_ssl = false
+is_test = true
+# CORS 허용 오리진(프론트). 비우면 [] (CORS 미적용). 5173=vite dev.
+cors_origins = ["http://localhost:5173", "http://127.0.0.1:5173"]
+
+[LogConfig]
+print_console = true
+log_level = "debug"
+
+# DB Read/Write 분리. 도커 실행 시 host 는 docker-compose 의 DB_HOST 로 override.
+# 관리형 DB(RDS/Aurora/Azure)는 host 에 엔드포인트, sslmode="require".
+# LPS 도메인 로직/테이블이 생기기 전까지는 접속하지 않으므로(엔진 lazy) placeholder 여도 부팅된다.
+[MainDBConfig]
+db_type = "postgresql"
+name = "lps_db"
+write_host = "127.0.0.1"
+write_port = 5432
+write_id = "<DB_USER>"
+write_pw = "<DB_PASSWORD>"
+read_host = "127.0.0.1"
+read_port = 5432
+read_id = "<DB_USER>"
+read_pw = "<DB_PASSWORD>"
+show_log = false
+pool_size = 10
+max_overflow = 20
+sslmode = ""              # 로컬: "" / 관리형 DB: "require"|"verify-ca"|"verify-full"

lps/config/config_loader.py

@@ -0,0 +1,37 @@
+import tomllib
+from typing import Optional, Type, Dict, TypeVar
+from pydantic import BaseModel
+
+
+class ConfigModel(BaseModel):
+    pass
+
+
+# APP_ENV
+# local : 로컬 환경(개인 pc)
+# dev   : 개발환경 (사내 pc)
+# prod  : 서비스 환경 (클라우드 서버)
+#
+# 실행시 환경변수 설정
+# linux  : export APP_ENV=dev
+# window : set APP_ENV=dev
+class Configs:
+    ConfigType = TypeVar("ConfigType", bound=ConfigModel)
+
+    def __init__(self, file_path: str):
+        self._settings: Dict[Type["Configs.ConfigType"], "Configs.ConfigType"] = self._load_settings_from_toml(file_path)
+
+    def _load_settings_from_toml(self, file_path: str) -> Dict[Type[ConfigType], ConfigType]:
+        with open(file_path, "rb") as f:
+            toml_content = tomllib.load(f)
+
+        config_subclasses = ConfigModel.__subclasses__()
+        configs = {
+            config_class: config_class.model_validate(toml_content[config_class.__name__])
+            for config_class in config_subclasses
+            if config_class.__name__ in toml_content
+        }
+        return configs
+
+    def get(self, config_class: Type[ConfigType]) -> Optional[ConfigType]:
+        return self._settings.get(config_class)

lps/config/config_models.py

@@ -0,0 +1,38 @@
+from config.config_loader import ConfigModel
+
+
+class WebServerConfig(ConfigModel):
+    server_name: str = ""
+    port: int = 0
+    process_count: int = 1
+    is_ssl: bool = False
+    is_test: bool = False
+    # CORS 허용 오리진(프론트). 비우면 CORS 미적용. 예: ["http://localhost:5173"]
+    cors_origins: list[str] = []
+
+
+class LogConfig(ConfigModel):
+    print_console: bool = True
+    log_level: str = "debug"
+
+
+# DB Read/Write 분리 설정.
+# 하나의 논리 DB 에 대해 write(주) / read(복제) 접속 정보를 각각 가진다.
+class MainDBConfig(ConfigModel):
+    db_type: str = "postgresql"
+    name: str = ""
+    write_host: str = ""
+    write_port: int = 5432
+    write_id: str = ""
+    write_pw: str = ""
+    read_host: str = ""
+    read_port: int = 5432
+    read_id: str = ""
+    read_pw: str = ""
+    show_log: bool = False
+    # 커넥션 풀 사이징. 실제 동시 커넥션 상한 = (pool_size + max_overflow) x 엔진수(R/W=2) x 워커수.
+    # PostgreSQL max_connections 를 넘지 않도록 설정해야 한다. (예: 10+20=30 x 2 x 5워커 = 300)
+    pool_size: int = 10
+    max_overflow: int = 20
+    # SSL/TLS 모드: ""/"disable"=미사용(로컬), "require"/"verify-ca"/"verify-full"=관리형 DB(RDS/Aurora/Azure).
+    sslmode: str = ""

lps/config/server_configs.py

@@ -0,0 +1,38 @@
+import os
+
+from config.config_loader import Configs
+from config.config_models import WebServerConfig, LogConfig, MainDBConfig
+
+# 실행 환경 결정 (기본 local). 환경변수 APP_ENV 로 변경.
+APP_ENV = os.environ.get("APP_ENV", "local")
+
+_config_dir = os.path.dirname(__file__)
+_config_file = os.path.join(_config_dir, f"config.{APP_ENV}.toml")
+
+# 운영 전제: 항상 APP_ENV=local 로 띄운다 → config.local.toml 사용 (test/docker 도 local 로 실행).
+if not os.path.exists(_config_file):
+    raise FileNotFoundError(f"설정 파일이 없습니다: {_config_file} (APP_ENV={APP_ENV}). APP_ENV=local 로 실행하세요.")
+
+configs = Configs(_config_file)
+
+web_server_config: WebServerConfig = configs.get(WebServerConfig)
+log_config: LogConfig = configs.get(LogConfig)
+main_db_config: MainDBConfig = configs.get(MainDBConfig)
+
+
+# DB 접속 env override (config.local.toml 유지, 도커에서 host 만 교체). 로컬은 env 미설정 → toml 그대로.
+def _apply_db_env_override(cfg: MainDBConfig):
+    h = os.environ.get("DB_HOST")
+    if h:
+        cfg.write_host = cfg.read_host = h
+    if os.environ.get("DB_PORT"):
+        cfg.write_port = cfg.read_port = int(os.environ["DB_PORT"])
+    if os.environ.get("DB_USER"):
+        cfg.write_id = cfg.read_id = os.environ["DB_USER"]
+    if os.environ.get("DB_PASSWORD"):
+        cfg.write_pw = cfg.read_pw = os.environ["DB_PASSWORD"]
+    if os.environ.get("DB_NAME"):
+        cfg.name = os.environ["DB_NAME"]
+
+
+_apply_db_env_override(main_db_config)

lps/conftest.py

@@ -0,0 +1,55 @@
+# 테스트도 APP_ENV=local 로 실행한다 (config.local.toml 사용).
+# config.server_configs 가 import 되는 순간 config.<APP_ENV>.toml 을 읽으므로 가장 먼저 설정.
+import os
+
+os.environ.setdefault("APP_ENV", "local")
+
+import pytest_asyncio
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from common.database.model.models import MAIN_BASE
+from config.server_configs import main_db_config
+
+
+def _write_url(cfg) -> str:
+    pw = f":{cfg.write_pw}" if cfg.write_pw else ""
+    return f"postgresql+asyncpg://{cfg.write_id}{pw}@{cfg.write_host}:{cfg.write_port}/{cfg.name}"
+
+
+@pytest_asyncio.fixture
+async def db_engine():
+    """테스트용 스키마를 보장한다(실제 DB 필요). 도메인 테이블이 생기면 이 fixture 를 쓰는 테스트를 추가한다.
+
+    앱(DB_SESSION_MNG)은 자체 엔진으로 같은 DB 에 접속하므로 여기서 만든 스키마를 그대로 공유한다.
+    """
+    engine = create_async_engine(_write_url(main_db_config))
+    async with engine.begin() as conn:
+        await conn.run_sync(MAIN_BASE.metadata.create_all)  # 이미 있으면 skip
+    yield engine
+    await engine.dispose()
+
+
+@pytest_asyncio.fixture(scope="session", autouse=True)
+async def _dispose_app_engines():
+    """테스트 세션이 끝날 때 앱 싱글톤 엔진을 정리한다.
+    (이벤트 루프 종료 후 커넥션이 GC 되며 나오는 'Event loop is closed' 경고 제거)
+    """
+    yield
+    from common.database.db_session_manager import DB_SESSION_MNG
+
+    await DB_SESSION_MNG.dispose_all()
+
+
+@pytest_asyncio.fixture
+async def client():
+    """앱을 실제 네트워크 없이 호출하는 httpx 클라이언트 (ASGITransport).
+
+    아직 도메인 테이블이 없어 DB 없이도 부팅되므로 db_engine 에 의존하지 않는다.
+    DB 를 쓰는 도메인 테스트를 추가할 땐 인자에 db_engine 을 받아 스키마를 보장한다.
+    """
+    from router.router import app
+
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as ac:
+        yield ac

lps/crud/.gitkeep

@@ -0,0 +1,2 @@
+# crud 폴더 placeholder — DB 접근 계층(crud)을 여기에 추가한다.
+# backend/crud/*.py 컨벤션: ISomethingCRUD(ABC) + SomethingCRUD 구현, DB_SESSION_MNG.execute 등 저수준 연산 호출.

lps/pytest.ini

@@ -0,0 +1,9 @@
+[pytest]
+asyncio_mode = auto
+# DB_SESSION_MNG(싱글톤)의 커넥션 풀이 첫 이벤트 루프에 묶이므로,
+# 모든 테스트/픽스처가 단일 session 루프를 공유하게 한다.
+asyncio_default_fixture_loop_scope = session
+asyncio_default_test_loop_scope = session
+testpaths = tests
+filterwarnings =
+    ignore::pytest.PytestUnraisableExceptionWarning

lps/requirements.txt

@@ -0,0 +1,8 @@
+fastapi
+uvicorn[standard]
+sqlalchemy>=2.0
+greenlet  # SQLAlchemy async 의 sync/async 브리지에 필수 (일부 환경에서 자동 설치 누락됨)
+asyncpg
+orjson
+pydantic>=2.0
+httpx  # 외부 사이트/오픈API(최저가 조회) 호출용 async HTTP 클라이언트

lps/router/router.py

@@ -0,0 +1,65 @@
+import time
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.middleware.gzip import GZipMiddleware
+
+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.<domain>.<file>
+#   app.include_router(router.v1.<domain>.<file>.router)
+
+API_SERVER_START_TIME = GTime.UTCStr()
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # startup
+    yield
+    # shutdown: DB 엔진 커넥션 풀 정리
+    await DB_SESSION_MNG.dispose_all()
+
+
+app = FastAPI(title="LPS Api Server", lifespan=lifespan)
+
+# CORS: config 의 cors_origins 가 있을 때만 적용(브라우저 프론트 호출 허용).
+# 명시적 오리진을 쓰므로 allow_credentials=True 가능(쿠키/Authorization 헤더 허용).
+if web_server_config.cors_origins:
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=web_server_config.cors_origins,
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+# Accept-Encoding: gzip 요청에 대해 1000 bytes 이상 응답을 압축.
+app.add_middleware(GZipMiddleware, minimum_size=1000)
+
+
+@app.middleware("http")
+async def log_time(request: Request, call_next):
+    start_time = time.time()
+    response = await call_next(request)
+    elapsed = time.time() - start_time
+    LOG.d(f"took: {elapsed:.4f} - {request.url.path}")
+    return response
+
+
+@app.get(
+    path="/healthz",
+    summary="헬스체크",
+    description="서버 기동 시각(API_SERVER_START_TIME)을 반환하는 헬스체크 엔드포인트.",
+    responses={404: {"description": "Not found"}},
+)
+async def healthz():
+    return API_SERVER_START_TIME
+
+
+# 각 도메인 라우터를 등록한다. 새 기능 추가 시 router.v1.<domain>.<file> 를 import 후 include.
+# (아직 도메인 로직 미정 — healthz 만 노출)

lps/router/v1/validator/dependencies.py

@@ -0,0 +1,20 @@
+from typing import Any
+
+from fastapi.responses import JSONResponse
+
+
+# ---- ResponseNone 처리 -----------------------------------------------------
+# 응답 객체에서 값이 None 인 필드를 재귀적으로 제거하여 페이로드를 줄인다.
+# 모든 라우터는 return RemoveNoneResponse(await service....) 형태로 반환한다.
+def RemoveNoneValues(obj: Any) -> Any:
+    if isinstance(obj, dict):
+        return {k: RemoveNoneValues(v) for k, v in obj.items() if v is not None}
+    if isinstance(obj, list):
+        return [RemoveNoneValues(v) for v in obj]
+    return obj
+
+
+def RemoveNoneResponse(obj) -> JSONResponse:
+    # mode="json": datetime/uuid 등을 JSON-safe 문자열로 변환(표준 JSONResponse 가 직렬화 가능).
+    # (ORJSONResponse 는 최신 FastAPI 에서 deprecated)
+    return JSONResponse(content=RemoveNoneValues(obj.model_dump(mode="json")))

lps/run_local_server.sh

@@ -0,0 +1,57 @@
+#!/usr/bin/env bash
+#
+# 로컬 LPS 서버 실행 (대화형). 실행하면 모드를 골라 입력한다.
+# 최초 실행 시 venv 생성 + 의존성 설치까지 자동으로 한다.
+#
+set -euo pipefail
+cd "$(dirname "$0")"   # lps/
+
+VENV=".venv"
+PY="$VENV/bin/python"
+PORT=9400
+
+# 1) venv + 의존성 보장
+if [[ ! -d "$VENV" ]]; then
+  echo "[setup] venv 생성 + 의존성 설치..."
+  python3 -m venv "$VENV"
+  "$PY" -m pip install -q --upgrade pip
+  "$PY" -m pip install -q -r requirements.txt
+fi
+
+# 2) config 보장
+if [[ ! -f config/config.local.toml ]]; then
+  echo "[error] config/config.local.toml 이 없습니다. 아래로 생성 후 값을 채우세요:"
+  echo "        cp config/config.local.toml.example config/config.local.toml"
+  exit 1
+fi
+
+# 3) 모드 선택
+echo "── 실행 모드 선택 ──"
+echo "  1) 일반 실행 (web_main.py)"
+echo "  2) 자동 재시작 (uvicorn --reload, 개발용)"
+echo "  3) 의존성 재설치"
+echo "  q) 취소"
+read -rp "선택 [1]: " choice
+choice="${choice:-1}"
+
+case "$choice" in
+  3) echo "[setup] 의존성 재설치..."; "$PY" -m pip install -q -r requirements.txt; echo "완료"; exit 0 ;;
+  q|Q) echo "취소합니다."; exit 0 ;;
+esac
+
+# 4) 포트 정리 (이미 떠 있으면 종료)
+if lsof -ti:"$PORT" >/dev/null 2>&1; then
+  echo "[info] 포트 $PORT 사용 중 → 기존 프로세스 종료"
+  lsof -ti:"$PORT" | xargs kill 2>/dev/null || true
+  sleep 1
+fi
+export APP_ENV=local
+
+# 5) 실행
+case "$choice" in
+  1) echo "[run] web_main.py  →  http://localhost:$PORT/docs"
+     exec "$PY" web_main.py ;;
+  2) echo "[run] uvicorn --reload  →  http://localhost:$PORT/docs"
+     exec "$VENV/bin/uvicorn" router.router:app --host 0.0.0.0 --port "$PORT" --reload ;;
+  *) echo "[error] 알 수 없는 선택: $choice"; exit 1 ;;
+esac

lps/services/.gitkeep

@@ -0,0 +1,2 @@
+# services 폴더 placeholder — 도메인 비즈니스 로직(service)을 여기에 추가한다.
+# backend/services/*.py 컨벤션: 라우터가 Depends 로 주입받고, DB 접근은 crud + DB_SESSION_MNG 람다로 위임.

lps/tests/test_health.py

@@ -0,0 +1,9 @@
+# 프레임워크 골격 스모크 테스트. 도메인 로직이 없어도 앱이 부팅되고 healthz 가 응답하는지 확인한다.
+# (DB 없이 통과 — 엔진은 lazy 라 실제 커넥션을 맺지 않는다.)
+
+
+async def test_healthz_ok(client):
+    res = await client.get("/healthz")
+    assert res.status_code == 200
+    # 기동 시각 문자열(예: "2026-07-08 04:26:58")을 그대로 반환한다.
+    assert isinstance(res.json(), str)

lps/web_main.py

@@ -0,0 +1,41 @@
+# 실행 방법
+#   pip install -r requirements.txt
+#   python web_main.py                # 기본 local 환경
+#   APP_ENV=dev python web_main.py    # 환경 지정
+#
+# 또는 uvicorn 직접 실행:
+#   uvicorn router.router:app --reload --host=0.0.0.0 --port=9400
+
+import uvicorn
+
+from common.logger import LOG
+from config.server_configs import web_server_config
+
+LOG.SetPrefix(web_server_config.server_name)
+
+# import 시점에 app 및 DB 세션 매니저(싱글톤)가 초기화된다.
+import router.router
+
+if __name__ == "__main__":
+    LOG.i(f"Server Name : {web_server_config.server_name}")
+    LOG.i(f"Server Port : {web_server_config.port}")
+    LOG.i(f"API Server start time : {router.router.API_SERVER_START_TIME}")
+
+    if web_server_config.is_ssl:
+        uvicorn.run(
+            "router.router:app",
+            host="0.0.0.0",
+            port=web_server_config.port,
+            access_log=False,
+            workers=web_server_config.process_count,
+            ssl_keyfile="./SSL/key.pem",
+            ssl_certfile="./SSL/cert.pem",
+        )
+    else:
+        uvicorn.run(
+            "router.router:app",
+            host="0.0.0.0",
+            port=web_server_config.port,
+            access_log=False,
+            workers=web_server_config.process_count,
+        )