import asyncio
import logging
from dataclasses import dataclass
from .db import session_scope
from .models import Order
logger = logging.getLogger(__name__)
RETRY_LIMIT = 3
@dataclass
class SyncResult:
total: int = 0
ok: int = 0
failed: int = 0
async def sync_batch(batch_id: str, limit: int = 500) -> SyncResult:
"""按批次同步订单,失败项进入重试队列。"""
result = SyncResult()
async with session_scope() as db:
rows = await db.fetch_all(
"SELECT id, payload FROM order_queue "
"WHERE batch_id = $1 ORDER BY id LIMIT $2",
batch_id, limit,
)
for row in rows:
try:
await _apply(db, row)
result.ok += 1
except Exception as exc:
logger.warning(
"apply failed id=%s err=%s", row["id"], exc)
result.failed += 1
result.total += 1
logger.info("batch %s done: %s", batch_id, result)
return result
async def _apply(db, row) -> None:
order = Order.from_payload(row["payload"])
if order.status == "pending":
await db.execute(
"UPDATE order_queue SET status='done' WHERE id=$1", row["id"])
await db.commit()