Класс коннектор для Диадок API на Python
Решил поделиться своим опытом как я собирался сделать сервис управлением ЭДО провайдерами по правилам SOLID.
Для начала я решил составить архитектуру сервиса, решил что класс управления api должен включать в себя http клиент как зависимость, так как не все могут захотеть использовать requests для выполнения запросов, еще это даст возможность переехать на асинхронную версию. Изучив документацию системы Диадок, я узнал что запросы можно выполнять как в JSON формате так и используя RPC модели. Поэтому я назвал класс DiadocJSONClient и он использует библиотеку requests для http запросов.
class DiadocJSONClient:
"""Клиент АПИ запросов."""
session = None
response_obj = RequestsResponse
def __init__(
self,
url: str,
login: str = None,
password: str = None,
api_client_id: str = None,
):
self.url = url
self._login = login
self._password = password
self._api_client_id = api_client_id
def __enter__(self):
logger.info("Create client connection")
created, self.session = self._session_get_or_create()
return self
def __exit__(self, exc_type, exc_value, traceback):
logger.info("Close client connection!")
self._close_session()
def _create_session(self):
"""Создать сессию."""
token = self._get_token()
headers = self._get_headers(token)
session = requests.Session()
session.headers.update(headers)
logger.success("Session created!")
return session
def _close_session(self):
"""Закрыть сессию."""
self._check_session_is_exists()
self.session.close()
def _session_get_or_create(self):
if not self.session:
logger.info("SESSION NOT FIND!")
return True, self._create_session()
return False, self.session
def _get_token(self) -> str:
"""Получить токен."""
url = f"{self.url}/V3/Authenticate"
auth_str = f"DiadocAuth ddauth_api_client_id={self._api_client_id}"
headers = {"Authorization": auth_str}
params = {"type": "password"}
body = {"login": self._login, "password": self._password}
try:
response = requests.post(
url, headers=headers, params=params, json=body
)
response.raise_for_status()
except Exception as err:
logger.error("{}: {}", err.__class__.__name__, err)
raise TokenReceiptError(
"Ошибка получения токена: {}".format(err.__class__.__name__)
)
token = response.text
return token
def _get_headers(self, token: str) -> dict:
"""Получить headers."""
auth_str = (
"DiadocAuth "
f"ddauth_api_client_id={self._api_client_id}, "
f"ddauth_token={token}"
)
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
"Authorization": auth_str,
}
return headers
def post(self, method, body=None, params=None) -> HTTPResponse:
"""POST запрос."""
created, session = self._session_get_or_create()
body = body or {}
params = params or {}
request_kwargs = {"params": params, "json": body}
url = f"{self.url}/{method}"
logger.debug("POST /{}, body={}, params={}", method, body, params)
try:
response = session.post(url, **request_kwargs)
logger.debug(response.request.headers)
response.raise_for_status()
except Exception as err:
logger.error("{}: {}", err.__class__.__name__, err)
try:
error_text = response.text
logger.debug(error_text)
except Exception:
error_text = ""
# logger.debug(response.request.body)
raise RequestError(
f"Ошибка выполнения запроса: "
f"POST /{method}: {err}, "
f"text: {error_text}"
)
if created:
session.close()
logger.debug("{}: {}", response.status_code, response.text)
return self.response_obj(response)
def post_binary(
self,
method,
params=None,
files_content: bytes = None,
):
"""POST запрос."""
created, session = self._session_get_or_create()
params = params or {}
url = f"{self.url}/{method}"
logger.debug("POST BINARY /{}, params={}", method, params)
try:
response = session.post(url, params=params, data=files_content)
logger.debug(response.request.headers)
response.raise_for_status()
except Exception as err:
logger.error("{}: {}", err.__class__.__name__, err)
try:
error_text = response.text
logger.debug(error_text)
except Exception:
error_text = ""
# logger.debug(response.request.body)
raise RequestError(
f"Ошибка выполнения запроса: "
f"POST /{method}: {err}, "
f"text: {error_text}"
)
if created:
session.close()
logger.debug("{}: {}", response.status_code, response.text)
return self.response_obj(response)
def get(self, method, params=None, headers=None) -> HTTPResponse:
"""GET запрос."""
created, session = self._session_get_or_create()
params = params or {}
headers = headers or {}
session.headers.update(headers)
url = f"{self.url}/{method}"
logger.debug("GET /{}, params={}", url, params)
try:
response = session.get(url, params=params)
logger.debug(response.request.headers)
response.raise_for_status()
except Exception as err:
logger.error("{}: {}", err.__class__.__name__, err)
raise RequestError(
f"Ошибка выполнения запроса: GET /{method}: {err}"
)
if created:
session.close()
logger.debug("response: {}", response.text[:200])
return self.response_obj(response)Немного расскажу об основных методах класса
__init__ — принимает креды для авторизации
get — выполняет GET запрос к Диадоку
post — выполняет POST запрос к Диадоку. Изначально в методе (как и в get) проверяется есть ли открытая авторизованная сессия, если нет то создается сессия в методе класса. Это сделано для того чтобы можно было выполнять несколько запросов получив токен один раз на всю сессию. если запрос создается вне сессии то сессия создастся на этот запрос и закроется после выполнения метода.
Диадок выдает не всю информацию в теле ответа, иногда некоторые параметры передаются в headers ответа. Ввиду этого мне пришлось сделать класс для ответов клиента, чтобы независимо от используемой библиотеки для запросов, ответ должен был иметь одинаковые методы.
Я создал свойство response_obj = RequestsResponse
Интерфейс модели ответа
from abc import ABC, abstractmethod
from typing import Any
class HTTPResponse(ABC):
@abstractmethod
def __init__(self, response: Any):
self._response = response
@property
@abstractmethod
def status_code(self) -> int:
pass
@property
@abstractmethod
def headers(self) -> dict[str, str]:
pass
@property
@abstractmethod
def content(self) -> bytes:
pass
@property
@abstractmethod
def text(self) -> bytes:
pass
@abstractmethod
def json(self) -> Any:
pass
def raise_for_status(self) -> None:
pass
Сама модель ответа
class RequestsResponse(HTTPResponse):
def __init__(self, response: Response):
self._response = response
@property
def status_code(self) -> int:
return self._response.status_code
@property
def headers(self) -> dict[str, str]:
return dict(self._response.headers)
@property
def content(self) -> bytes:
return self._response.content
@property
def text(self) -> str:
return self._response.text
def json(self) -> dict:
return self._response.json()
def raise_for_status(self) -> None:
self._response.raise_for_status()Все ответы метода get и post привожу к свой общей модели,
return self.response_obj (response)
def get(self, method, params=None, headers=None) -> HTTPResponse:
"""GET запрос."""
created, session = self._session_get_or_create()
params = params or {}
headers = headers or {}
session.headers.update(headers)
url = f"{self.url}/{method}"
logger.debug("GET /{}, params={}", url, params)
try:
response = session.get(url, params=params)
logger.debug(response.request.headers)
response.raise_for_status()
except Exception as err:
logger.error("{}: {}", err.__class__.__name__, err)
raise RequestError(
f"Ошибка выполнения запроса: GET /{method}: {err}"
)
if created:
session.close()
logger.debug("response: {}", response.text[:200])
return self.response_obj(response)Тут я описал структуру моего клиента для запросов к API Диадок. В следующей статья я опишу как этот клиент встраивается в класс провайдера, который уже непосредственно выполняет запросы и обрабатывает ответы.
