49 lines
2.1 KiB
Python
49 lines
2.1 KiB
Python
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()
|