- All frontend pages, labels, notices and errors; html lang=uk, uk-UA money format - Brand "Assistant System" in the top bar and page title - Backend error details returned to the UI (auth, orders, receipts, cash registers, Checkbox/CRM/NP errors) and CLI output - Tests updated for the new messages Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
74 lines
2.7 KiB
Python
74 lines
2.7 KiB
Python
"""Protocol клиента Checkbox — позволяет подменять реализацию в тестах и локально.
|
|
|
|
См. `StubCheckboxClient`. Выбор реализации — `get_checkbox_client()` ниже,
|
|
единственное место, читающее `CHECKBOX_USE_STUB`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from functools import lru_cache
|
|
from typing import Any, Protocol
|
|
|
|
from app.core.config import settings
|
|
from app.schemas.checkbox import EttnOut
|
|
|
|
|
|
class CheckboxError(Exception):
|
|
"""Checkbox отклонил запрос (4xx) — повтор с тем же телом не поможет."""
|
|
|
|
|
|
class CheckboxUnavailableError(CheckboxError):
|
|
"""Сеть, таймаут или 5xx — исход запроса неизвестен, можно повторить."""
|
|
|
|
|
|
class CheckboxRateLimitedError(CheckboxUnavailableError):
|
|
"""Лимит частоты запросов (Checkbox 429 или «To many requests» от Новой Почты).
|
|
|
|
В отличие от прочих `CheckboxUnavailableError` исход известен — запрос отклонён,
|
|
повторять можно не раньше чем через `retry_after` секунд.
|
|
"""
|
|
|
|
def __init__(self, message: str, retry_after: float) -> None:
|
|
super().__init__(message)
|
|
self.retry_after = retry_after
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CheckboxCredentials:
|
|
"""Расшифрованные доступы одной кассы. Живут только в памяти."""
|
|
|
|
cash_register_id: uuid.UUID
|
|
license_key: str
|
|
pin_code: str
|
|
|
|
|
|
class CheckboxClient(Protocol):
|
|
async def sign_in(self, creds: CheckboxCredentials) -> None: ...
|
|
|
|
async def create_ettn(self, creds: CheckboxCredentials, body: dict[str, Any]) -> EttnOut: ...
|
|
|
|
async def get_ettn(self, creds: CheckboxCredentials, ettn_id: str) -> EttnOut: ...
|
|
|
|
async def find_ettn(
|
|
self, creds: CheckboxCredentials, waybill_number: str
|
|
) -> EttnOut | None: ...
|
|
|
|
async def delete_ettn(self, creds: CheckboxCredentials, ettn_id: str) -> None: ...
|
|
|
|
|
|
@lru_cache
|
|
def get_checkbox_client() -> CheckboxClient:
|
|
"""Один экземпляр на процесс: внутри — кэш токенов кассиров."""
|
|
if settings.checkbox_use_stub:
|
|
if settings.is_production:
|
|
raise RuntimeError("CHECKBOX_USE_STUB=true заборонено в production")
|
|
from app.services.checkbox.stub_client import StubCheckboxClient
|
|
|
|
return StubCheckboxClient(auto_complete_after=2)
|
|
|
|
from app.services.checkbox.http_client import HttpCheckboxClient
|
|
|
|
return HttpCheckboxClient(settings)
|