full peresobral
This commit is contained in:
121
sd_api.py
Normal file
121
sd_api.py
Normal file
@@ -0,0 +1,121 @@
|
||||
# sd_api.py
|
||||
|
||||
import requests
|
||||
import logging
|
||||
import json
|
||||
from typing import List, Dict, Any
|
||||
|
||||
# Импортируем нашу конфигурацию
|
||||
import config
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
class ServiceDeskAPIError(Exception):
|
||||
"""Кастомное исключение для ошибок API ServiceDesk."""
|
||||
pass
|
||||
|
||||
class ServiceDeskClient:
|
||||
"""
|
||||
Клиент для взаимодействия с REST API ServiceDesk.
|
||||
"""
|
||||
def __init__(self, access_key: str, base_url: str):
|
||||
if not access_key or not base_url:
|
||||
raise ValueError("Access key and base URL cannot be empty.")
|
||||
|
||||
self.base_url = base_url
|
||||
self.session = requests.Session()
|
||||
# Устанавливаем accessKey как параметр по умолчанию для всех запросов
|
||||
self.session.params = {'accessKey': access_key}
|
||||
|
||||
def _make_request(self, method: str, url: str, params: Dict = None, json_data: Dict = None) -> Any:
|
||||
"""
|
||||
Внутренний метод для выполнения HTTP-запросов.
|
||||
|
||||
:param method: HTTP метод ('GET', 'POST', etc.)
|
||||
:param url: Полный URL для запроса.
|
||||
:param params: Дополнительные параметры URL (кроме accessKey).
|
||||
:param json_data: Тело запроса в формате JSON для POST/PUT.
|
||||
:return: Ответ сервера в виде JSON.
|
||||
:raises ServiceDeskAPIError: в случае ошибки API или сети.
|
||||
"""
|
||||
try:
|
||||
response = self.session.request(method, url, params=params, json=json_data, timeout=30)
|
||||
response.raise_for_status() # Вызовет исключение для кодов 4xx/5xx
|
||||
|
||||
# Некоторые ответы могут быть пустыми (например, при успешном редактировании)
|
||||
if response.status_code == 204 or not response.text:
|
||||
return None
|
||||
|
||||
return response.json()
|
||||
|
||||
except requests.exceptions.HTTPError as e:
|
||||
error_message = f"HTTP Error: {e.response.status_code} for URL {url}. Response: {e.response.text}"
|
||||
log.error(error_message)
|
||||
raise ServiceDeskAPIError(error_message) from e
|
||||
except requests.exceptions.RequestException as e:
|
||||
error_message = f"Request failed for URL {url}: {e}"
|
||||
log.error(error_message)
|
||||
raise ServiceDeskAPIError(error_message) from e
|
||||
|
||||
def get_all_frs(self) -> List[Dict]:
|
||||
"""Получает список всех фискальных регистраторов."""
|
||||
log.info("Запрос списка всех ФР из ServiceDesk...")
|
||||
params = {
|
||||
'attrs': 'UUID,FRSerialNumber,RNKKT,KKTRegDate,FNExpireDate,FNNumber,owner,FRDownloader,LegalName,OFDName,ModelKKT,FFD,lastModifiedDate'
|
||||
}
|
||||
frs = self._make_request('POST', config.FIND_FRS_URL, params=params)
|
||||
log.info(f"Получено {len(frs)} записей о ФР.")
|
||||
return frs
|
||||
|
||||
def get_all_workstations(self) -> List[Dict]:
|
||||
"""Получает список всех рабочих станций."""
|
||||
log.info("Запрос списка всех рабочих станций из ServiceDesk...")
|
||||
params = {
|
||||
'attrs': 'UUID,owner,AnyDesk,Teamviewer'
|
||||
}
|
||||
workstations = self._make_request('POST', config.FIND_WORKSTATIONS_URL, params=params)
|
||||
log.info(f"Получено {len(workstations)} записей о рабочих станциях.")
|
||||
return workstations
|
||||
|
||||
def get_lookup_values(self, metaclass: str) -> List[Dict]:
|
||||
"""
|
||||
Получает значения из справочника по его metaClass.
|
||||
|
||||
:param metaclass: metaClass справочника (e.g., 'ModeliFR', 'FFD').
|
||||
:return: Список словарей с 'UUID' и 'title'.
|
||||
"""
|
||||
log.info(f"Запрос значений справочника для metaClass: {metaclass}...")
|
||||
url = f"{self.base_url}/find/{metaclass}"
|
||||
params = {'attrs': 'UUID,title'}
|
||||
lookups = self._make_request('POST', url, params=params)
|
||||
log.info(f"Получено {len(lookups)} значений для {metaclass}.")
|
||||
return lookups
|
||||
|
||||
def update_fr(self, uuid: str, data: Dict) -> None:
|
||||
"""
|
||||
Обновляет существующий фискальный регистратор.
|
||||
|
||||
:param uuid: UUID объекта для редактирования.
|
||||
:param data: Словарь с полями для обновления.
|
||||
"""
|
||||
log.info(f"Обновление объекта ФР с UUID: {uuid}...")
|
||||
# Примечание: старый код использовал POST с параметрами для редактирования.
|
||||
# Если API требует form-encoded data, а не JSON, нужно использовать `data=data` вместо `json=data`.
|
||||
# Судя по вашему коду, это POST-запрос с параметрами в URL, а не в теле.
|
||||
url = config.EDIT_FR_URL_TEMPLATE.format(uuid=uuid)
|
||||
self._make_request('POST', url, params=data)
|
||||
log.info(f"Объект {uuid} успешно обновлен.")
|
||||
|
||||
def create_fr(self, data: Dict) -> Dict:
|
||||
"""
|
||||
Создает новый фискальный регистратор.
|
||||
|
||||
:param data: Словарь с данными для создания объекта (тело запроса).
|
||||
:return: JSON-ответ от сервера, содержащий UUID нового объекта.
|
||||
"""
|
||||
log.info(f"Создание нового ФР с серийным номером: {data.get('FRSerialNumber')}...")
|
||||
# Параметр для получения UUID в ответе
|
||||
params = {'attrs': 'UUID'}
|
||||
response = self._make_request('POST', config.CREATE_FR_URL, params=params, json_data=data)
|
||||
log.info(f"ФР успешно создан. Ответ сервера: {response}")
|
||||
return response
|
||||
Reference in New Issue
Block a user