feat(negodata): lps_db 읽기전용 연결 기반 — DBType.LPS + 조건부 엔진 등록

LPS(인터넷 최저가 검색) 연동 1단계. price_history 동기화 배치가 쓸
읽기전용 접속을 기존 확장점(DBType enum + 세션매니저 맵)에 등록한다.

- DBType.LPS=2, LpsDBConfig(read_* 만 — write 엔진 미등록으로 앱 차원
  읽기전용 강제, write 요청 시 KeyError)
- config name 비면 미등록(구버전 toml·LPS 없는 환경에서도 부팅 가능),
  배치는 is_registered() 로 사전 확인
- env override: LPS_DB_HOST/PORT/USER/PASSWORD/NAME (도커용)
- 검증: 엔진 등록·읽기 세션·write 가드 KeyError·lps_db.price_history
  실조회 확인

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

negodata/backend/common/database/db_session_manager.py

@@ -9,7 +9,7 @@ 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
+from config.server_configs import main_db_config, lps_db_config
 
 
 class DBSessionManager(Singleton):
@@ -47,6 +47,12 @@ class DBSessionManager(Singleton):
             DBType.MAIN.value: self.create_engine(DBType.MAIN.value, DBWRType.DB_READ.value),
         }
 
+        # LPS 결과 DB(lps_db) — **읽기전용**: read 엔진만 등록한다(write 로 열면 KeyError = 앱 차원 가드).
+        # config 의 name 이 비어 있으면 미사용(스키마 없는 환경에서도 부팅 가능, 배치는 is_registered 로 스킵).
+        if lps_db_config.name:
+            self.__db_type_map[DBType.LPS.value] = lps_db_config
+            self.__read_session[DBType.LPS.value] = self.create_engine(DBType.LPS.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:
@@ -83,6 +89,10 @@ class DBSessionManager(Singleton):
         )
         return scoped_session
 
+    def is_registered(self, db_type: int) -> bool:
+        """해당 논리 DB 가 등록돼 있는지 — 선택 연동(LPS 등)의 배치가 실행 전 확인하는 용도."""
+        return db_type in self.__db_type_map
+
     async def dispose_all(self):
         """모든 엔진의 커넥션 풀을 정리한다. 앱 종료/테스트 종료 시 호출한다.
         호출하지 않으면 풀 커넥션이 이벤트 루프 종료 후 GC 되며 경고를 남긴다.

negodata/backend/common/enums.py

@@ -99,6 +99,7 @@ class DBType(Enum):
     """
 
     MAIN = 1
+    LPS = 2   # 인터넷 최저가 검색(lps_db) — 읽기전용(price_history 동기화 배치용, write 엔진 미등록)
 
 
 class DBWRType(Enum):

negodata/backend/config/config.local.toml.example

@@ -32,6 +32,20 @@ pool_size = 10
 max_overflow = 20
 sslmode = ""              # 로컬: "" / 관리형 DB: "require"|"verify-ca"|"verify-full"
 
+# 인터넷 최저가 검색(LPS) 결과 DB — 읽기전용(price_history 동기화 배치).
+# name 을 비우면 LPS 연동 비활성(엔진 미등록·배치 스킵). 도커는 LPS_DB_HOST 등 env 로 override.
+[LpsDBConfig]
+db_type = "postgresql"
+name = "lps_db"
+read_host = "127.0.0.1"
+read_port = 5432
+read_id = "postgres"
+read_pw = "postgres"
+show_log = false
+pool_size = 2
+max_overflow = 2
+sslmode = ""
+
 [JwtToken]
 access_key = "<JWT_ACCESS_SECRET>"
 refresh_key = "<JWT_REFRESH_SECRET>"

negodata/backend/config/config_models.py

@@ -39,6 +39,23 @@ class MainDBConfig(ConfigModel):
     sslmode: str = ""
 
 
+# 인터넷 최저가 검색(LPS)의 결과 DB(lps_db) — **읽기전용** 접속.
+# LPS 는 별도 서비스(자체 스키마 소유)이고, negodata 는 price_history 를 주기 배치로 읽어만 온다.
+# 그래서 read_* 만 둔다(write 엔진 미등록 = 앱 차원 읽기전용 강제). name 이 비면 미사용(등록 스킵).
+class LpsDBConfig(ConfigModel):
+    db_type: str = "postgresql"
+    name: str = ""            # 비우면 LPS 연동 비활성(엔진 미등록, 배치 스킵)
+    read_host: str = ""
+    read_port: int = 5432
+    read_id: str = ""
+    read_pw: str = ""
+    show_log: bool = False
+    # 배치 전용이라 작은 풀이면 충분 (동시 사용처 = 스케줄러 잡 1개)
+    pool_size: int = 2
+    max_overflow: int = 2
+    sslmode: str = ""
+
+
 class JwtToken(ConfigModel):
     access_key: str = ""
     refresh_key: str = ""

negodata/backend/config/server_configs.py

@@ -1,7 +1,7 @@
 import os
 
 from config.config_loader import Configs
-from config.config_models import WebServerConfig, LogConfig, MainDBConfig, JwtToken, StorageConfig, MailConfig
+from config.config_models import WebServerConfig, LogConfig, MainDBConfig, LpsDBConfig, JwtToken, StorageConfig, MailConfig
 
 # 실행 환경 결정 (기본 local). 환경변수 APP_ENV 로 변경.
 APP_ENV = os.environ.get("APP_ENV", "local")
@@ -18,6 +18,8 @@ configs = Configs(_config_file)
 web_server_config: WebServerConfig = configs.get(WebServerConfig)
 log_config: LogConfig = configs.get(LogConfig)
 main_db_config: MainDBConfig = configs.get(MainDBConfig)
+# [LpsDBConfig] 섹션이 없는 toml(구버전)에서도 죽지 않게 기본값 폴백(name 빈 값 → LPS 연동 비활성).
+lps_db_config: LpsDBConfig = configs.get(LpsDBConfig) or LpsDBConfig()
 jwt_token_config: JwtToken = configs.get(JwtToken)
 storage_config: StorageConfig = configs.get(StorageConfig)
 # [MailConfig] 섹션이 없는 toml(구버전)에서도 죽지 않도록 기본값으로 폴백(전 필드 빈 값 → 발송 시 EmailUnavailable).
@@ -40,3 +42,20 @@ def _apply_db_env_override(cfg: MainDBConfig):
 
 
 _apply_db_env_override(main_db_config)
+
+
+# LPS 결과 DB(읽기전용) env override — 도커에서 host 등만 교체. 로컬은 toml 그대로.
+def _apply_lps_db_env_override(cfg: LpsDBConfig):
+    if os.environ.get("LPS_DB_HOST"):
+        cfg.read_host = os.environ["LPS_DB_HOST"]
+    if os.environ.get("LPS_DB_PORT"):
+        cfg.read_port = int(os.environ["LPS_DB_PORT"])
+    if os.environ.get("LPS_DB_USER"):
+        cfg.read_id = os.environ["LPS_DB_USER"]
+    if os.environ.get("LPS_DB_PASSWORD"):
+        cfg.read_pw = os.environ["LPS_DB_PASSWORD"]
+    if os.environ.get("LPS_DB_NAME"):
+        cfg.name = os.environ["LPS_DB_NAME"]
+
+
+_apply_lps_db_env_override(lps_db_config)