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