diff --git a/backend/app/services/db_worker.py b/backend/app/services/db_worker.py new file mode 100644 index 0000000..fab4df2 --- /dev/null +++ b/backend/app/services/db_worker.py @@ -0,0 +1,48 @@ +import asyncio +import logging +from app.core.database import AsyncSessionLocal +from app.models.models import TradeHistory, StockAnalysis + +logger = logging.getLogger(__name__) + +# 전역(Global) 비동기 큐 생성 (메모리 버퍼 역할) +db_task_queue = asyncio.Queue() + +async def db_writer_worker(): + """ + 메인 스레드나 API 스레드를 막지 않고, + 백그라운드에서 큐에 쌓인 DB 작업만 순차적으로 처리하는 워커 + DB 커넥션 경합과 락(Lock)을 방지합니다. + """ + logger.info("🛠️ Background DB Writer Worker started") + while True: + try: + # 큐에 데이터가 들어올 때까지 대기 + task = await db_task_queue.get() + task_type = task.get("type") + data = task.get("data") + + # 워커 전용 DB 세션을 열어서 처리 (락 경합 최소화) + async with AsyncSessionLocal() as session: + if task_type == "insert_trade_history": + # 매매일지 추가 + new_trade = TradeHistory(**data) + session.add(new_trade) + await session.commit() + logger.info(f"✅ [Worker] 매매일지 비동기 저장 완료: {data.get('stock_code', 'Unknown')}") + + elif task_type == "upsert_stock_analysis": + # 종목 분석 결과 저장 (추후 batch_analysis 등에서 이관될 로직용 플레이스홀더) + # 실제 upsert 로직은 데이터 구조에 맞춰 구현 + logger.info(f"✅ [Worker] 종목분석 비동기 저장 처리 (Placeholder): {data.get('stock_code', 'Unknown')}") + + else: + logger.warning(f"⚠️ [Worker] 알 수 없는 작업 타입: {task_type}") + + except asyncio.CancelledError: + logger.info("⏹️ DB Writer Worker stopped") + break + except Exception as e: + logger.error(f"❌ [Worker] DB 작업 중 에러 발생: {e}", exc_info=True) + finally: + db_task_queue.task_done()