Cashier creates an ETTN receipt in Checkbox bound to the TTN with payment control; Checkbox fiscalizes it itself when the parcel is paid for. - cash_registers (Fernet-encrypted license key / PIN) and receipts tables - Checkbox HTTP client + stub (ETTN does not work on test registers) - two-phase create via ARQ job, timeout reconciliation, cron status polling - /receipts and /cash-registers API, audit records - dashboard: per-order and bulk create, prepayment, cancel; cash registers page Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
62 lines
2.1 KiB
Python
62 lines
2.1 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 — исход запроса неизвестен, можно повторить."""
|
|
|
|
|
|
@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)
|