Connector class for Diadoc API in Python
I decided to share my experience of how I was going to make a service for EDI management by providers according to SOLID rules.
To begin with, I decided to draw up the architecture of the service, I decided that the api control class should include the http client as a dependency, since not everyone may want to use requests to execute requests, and this will also make it possible to move to the asynchronous version. Having studied the documentation of the Diadoc system, I learned that requests can be executed both in JSON format and using the RPC model. So I named the class DiadocJSONClient and it uses the requests library for http requests.
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)I'll tell you a little about the main methods of the class
__init__ – accepts credentials for authorization
get – performs a GET request to Diadoc
post – performs a POST request to Diadoc. Initially, the method (as in get) checks whether there is an open authorized session; if not, then a session is created in the class method. This is done so that you can perform several requests by receiving the token once for the entire session. if a request is created outside of a session, then a session will be created for this request and will be closed after the method is executed.
Diadoc does not display all the information in the response body; sometimes some parameters are passed to the response headers. In view of this, I had to make a class for client responses so that, regardless of the library used for requests, the response should have the same methods.
I created a property response_obj = RequestsResponse
Response Model Interface
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
The response model itself
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()I bring all the responses of the get and post method to my general model,
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)Here I described the structure of my client for requests to the Diadoc API. In the next article I will describe how this client is built into the provider class, which directly executes requests and processes responses.
