Класс коннектор для Диадок 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 Диадок. В следующей статье я опишу как этот клиент встраивается в класс провайдера, который уже непосредственно выполняет запросы и обрабатывает ответы.