Andrey Smakotin
b1b56fbb2e
- Добавлен endpoint /test/<integration_id>/ для тестирования соединений - RecommerceService упрощён под реальное API (x-auth-token + store_url) - Кнопка "Проверить подключение" в UI с обработкой статусов - Миграция для удаления IntegrationConfig и обновления полей Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
103 lines
3.5 KiB
Python
103 lines
3.5 KiB
Python
"""
|
||
Сервис для работы с Recommerce API.
|
||
|
||
API документация:
|
||
- Запросы отправляются на домен магазина (store_url)
|
||
- Авторизация через заголовок x-auth-token
|
||
- POST данные передаются как form-data
|
||
- Ответы в JSON
|
||
"""
|
||
import requests
|
||
from typing import Tuple
|
||
|
||
|
||
class RecommerceService:
|
||
"""Сервис для работы с Recommerce API"""
|
||
|
||
def __init__(self, integration):
|
||
"""
|
||
Args:
|
||
integration: RecommerceIntegration instance
|
||
"""
|
||
self.integration = integration
|
||
|
||
def _get_headers(self) -> dict:
|
||
"""Заголовки для API запросов"""
|
||
token = self.integration.api_token or ''
|
||
# HTTP заголовки должны быть ASCII-совместимыми
|
||
return {
|
||
'x-auth-token': token.encode('ascii', 'ignore').decode('ascii'),
|
||
'Accept': 'application/json',
|
||
}
|
||
|
||
def _get_url(self, path: str) -> str:
|
||
"""Полный URL для API endpoint"""
|
||
base = self.integration.store_url.rstrip('/')
|
||
return f"{base}/api/v1/{path.lstrip('/')}"
|
||
|
||
def _request(self, method: str, path: str, **kwargs) -> Tuple[bool, dict, str]:
|
||
"""
|
||
HTTP запрос к API.
|
||
|
||
Returns:
|
||
(success, data, error_message)
|
||
"""
|
||
url = self._get_url(path)
|
||
try:
|
||
response = requests.request(
|
||
method,
|
||
url,
|
||
headers=self._get_headers(),
|
||
timeout=15,
|
||
**kwargs
|
||
)
|
||
|
||
if response.status_code in [200, 201, 204]:
|
||
if response.text:
|
||
return True, response.json(), ''
|
||
return True, {}, ''
|
||
else:
|
||
return False, {}, f"HTTP {response.status_code}: {response.text[:200]}"
|
||
|
||
except requests.exceptions.Timeout:
|
||
return False, {}, 'Таймаут соединения (15 сек)'
|
||
except requests.exceptions.ConnectionError:
|
||
return False, {}, 'Не удалось подключиться к серверу'
|
||
except Exception as e:
|
||
return False, {}, str(e)
|
||
|
||
def test_connection(self) -> Tuple[bool, str]:
|
||
"""
|
||
Проверить соединение с Recommerce API.
|
||
|
||
Returns:
|
||
(success, message)
|
||
"""
|
||
if not self.integration.store_url:
|
||
return False, 'Не указан URL магазина'
|
||
if not self.integration.api_token:
|
||
return False, 'Не указан API токен'
|
||
|
||
url = self.integration.store_url.rstrip('/') + '/api/v1/'
|
||
try:
|
||
response = requests.get(
|
||
url,
|
||
headers=self._get_headers(),
|
||
timeout=15
|
||
)
|
||
|
||
if response.status_code == 401:
|
||
return False, 'Неверный API токен'
|
||
if response.status_code == 403:
|
||
return False, 'Доступ запрещён'
|
||
|
||
# Любой ответ от сервера = соединение работает
|
||
return True, 'Соединение установлено'
|
||
|
||
except requests.exceptions.Timeout:
|
||
return False, 'Таймаут соединения (15 сек)'
|
||
except requests.exceptions.ConnectionError:
|
||
return False, 'Не удалось подключиться к серверу'
|
||
except Exception as e:
|
||
return False, str(e)
|