lauadminandClaude Opus 5.5
528a869657
Checkbox relays Nova Poshta's "To many requests" (20000401501) as a 4xx third_party.generic error, not a 429, so bursts of receipt requests ended up as failed receipts. Such responses are now CheckboxRateLimitedError: the receipt stays pending and the worker job is retried with arq.Retry after the "Try again after N seconds" delay plus backoff. NP timeouts relayed the same way are treated as unavailable (unknown outcome). The HTTP client also sends Checkbox requests one at a time with a CHECKBOX_MIN_REQUEST_INTERVAL_MS pause and signs the cashier in once for concurrent jobs. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
78 lines
3.4 KiB
Python
78 lines
3.4 KiB
Python
"""ARQ worker: опрос Nova Poshta и ЕТТН-чеки Checkbox.
|
|
|
|
Запускается отдельным процессом: `arq app.worker.WorkerSettings`.
|
|
- `create_ettn_receipt` — задача, которую ставит API после запроса кассира;
|
|
- `poll_np_statuses` — раз в минуту статусы ТТН по заказам без чека;
|
|
- `poll_receipts` — раз в минуту повтор зависших `pending`, статусы `created`-чеков
|
|
и повтор смены статуса заказа в CRM (PACKED), если CRM была недоступна.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from arq import Retry, cron, func
|
|
from arq.connections import RedisSettings
|
|
|
|
from app.core.config import settings
|
|
from app.core.logging import configure_logging, get_logger
|
|
from app.db.session import SessionFactory
|
|
from app.services import receipts as receipts_service
|
|
from app.services.checkbox.client import CheckboxRateLimitedError, get_checkbox_client
|
|
from app.services.crm.exo_client import ExoCrmClient
|
|
from app.services.nova_poshta.np_client import NpTrackingClient
|
|
from app.services.orders import sync_np_statuses
|
|
|
|
log = get_logger(__name__)
|
|
|
|
# Повторы `create_ettn_receipt` при лимите частоты: пауза retry_after + 5 с × номер попытки.
|
|
_RATE_LIMIT_BACKOFF = 5
|
|
_CREATE_ETTN_MAX_TRIES = 6
|
|
|
|
|
|
async def startup(ctx: dict[str, Any]) -> None:
|
|
configure_logging()
|
|
ctx["np_client"] = NpTrackingClient(settings)
|
|
ctx["checkbox_client"] = get_checkbox_client()
|
|
ctx["crm_client"] = ExoCrmClient(settings)
|
|
log.info("worker_starting", environment=settings.environment)
|
|
|
|
|
|
async def poll_np_statuses(ctx: dict[str, Any]) -> None:
|
|
async with SessionFactory() as session:
|
|
await sync_np_statuses(session, ctx["np_client"])
|
|
log.info("np_statuses_polled")
|
|
|
|
|
|
async def create_ettn_receipt(ctx: dict[str, Any], receipt_id: str) -> None:
|
|
async with SessionFactory() as session:
|
|
try:
|
|
await receipts_service.create_ettn_for_receipt(
|
|
session, ctx["checkbox_client"], uuid.UUID(receipt_id)
|
|
)
|
|
except CheckboxRateLimitedError as exc:
|
|
# Лимит частоты Новой Почты: повтор с нарастающей паузой. Когда попытки
|
|
# кончатся, `pending`-чек подберёт cron `poll_receipts`.
|
|
raise Retry(defer=exc.retry_after + _RATE_LIMIT_BACKOFF * ctx["job_try"]) from exc
|
|
# Чек принят Checkbox — сразу переводим заказ в CRM в PACKED.
|
|
await receipts_service.sync_crm_statuses(session, ctx["crm_client"])
|
|
|
|
|
|
async def poll_receipts(ctx: dict[str, Any]) -> None:
|
|
async with SessionFactory() as session:
|
|
await receipts_service.retry_pending_receipts(session, ctx["checkbox_client"])
|
|
await receipts_service.sync_ettn_statuses(session, ctx["checkbox_client"])
|
|
await receipts_service.sync_crm_statuses(session, ctx["crm_client"])
|
|
log.info("receipts_polled")
|
|
|
|
|
|
class WorkerSettings:
|
|
redis_settings = RedisSettings.from_dsn(settings.redis_url)
|
|
on_startup = startup
|
|
functions = [func(create_ettn_receipt, max_tries=_CREATE_ETTN_MAX_TRIES)]
|
|
cron_jobs = [
|
|
cron(poll_np_statuses, minute=set(range(60)), run_at_startup=True),
|
|
cron(poll_receipts, minute=set(range(60)), second=30),
|
|
]
|