85 lines
3.1 KiB
Python
85 lines
3.1 KiB
Python
|
import logging
|
||
|
|
import requests
|
||
|
|
from datetime import datetime
|
||
|
|
from json import JSONDecodeError
|
||
|
|
from pprint import pformat
|
||
|
|
|
||
|
|
from odoo.exceptions import UserError
|
||
|
|
|
||
|
|
_logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
def _get_nilvera_client(company, timeout_limit=None):
|
||
|
|
return NilveraClient(
|
||
|
|
environment=company.l10n_tr_nilvera_environment,
|
||
|
|
api_key=company.l10n_tr_nilvera_api_key,
|
||
|
|
timeout_limit=timeout_limit,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class NilveraClient:
|
||
|
|
def __init__(self, environment=None, api_key=None, timeout_limit=None):
|
||
|
|
self.is_production = environment and environment == 'production'
|
||
|
|
self.base_url = 'https://api.nilvera.com' if self.is_production else 'https://apitest.nilvera.com'
|
||
|
|
self.timeout_limit = min(timeout_limit or 10, 30)
|
||
|
|
|
||
|
|
self.__session = requests.Session()
|
||
|
|
self.__session.headers.update({'Accept': 'application/json'})
|
||
|
|
if api_key:
|
||
|
|
self.__session.headers['Authorization'] = 'Bearer ' + api_key
|
||
|
|
|
||
|
|
def __enter__(self):
|
||
|
|
return self
|
||
|
|
|
||
|
|
def __exit__(self, type, value, traceback):
|
||
|
|
if hasattr(self, '_NilveraClient__session'):
|
||
|
|
self.__session.close()
|
||
|
|
|
||
|
|
def request(self, method, endpoint, params=None, json=None, files=None, handle_response=True):
|
||
|
|
start = datetime.utcnow()
|
||
|
|
url = self.base_url + endpoint
|
||
|
|
|
||
|
|
try:
|
||
|
|
response = self.__session.request(
|
||
|
|
method, url,
|
||
|
|
timeout=self.timeout_limit,
|
||
|
|
params=params,
|
||
|
|
json=json,
|
||
|
|
files=files,
|
||
|
|
)
|
||
|
|
except requests.exceptions.RequestException as e:
|
||
|
|
_logger.error("Network error during request: %s", e)
|
||
|
|
raise UserError("Network connectivity issue. Please check your internet connection and try again.")
|
||
|
|
|
||
|
|
end = datetime.utcnow()
|
||
|
|
self._log_request(method, start, end, url, params, json, response)
|
||
|
|
|
||
|
|
if handle_response:
|
||
|
|
return self.handle_response(response)
|
||
|
|
return response
|
||
|
|
|
||
|
|
def _log_request(self, method, start, end, url, params, json, response):
|
||
|
|
_logger.info(
|
||
|
|
"%(method)s\nstart=%(start)s\nend=%(end)s\nurl=%(url)s\nparams=%(params)s\njson=%(json)s\nresponse=%(response)s",
|
||
|
|
{
|
||
|
|
"method": method,
|
||
|
|
"start": start,
|
||
|
|
"end": end,
|
||
|
|
"url": pformat(url),
|
||
|
|
"params": pformat(params),
|
||
|
|
"json": pformat(json),
|
||
|
|
"response": pformat(response),
|
||
|
|
},
|
||
|
|
)
|
||
|
|
|
||
|
|
def handle_response(self, response):
|
||
|
|
if response.status_code in {401, 403}:
|
||
|
|
raise UserError("Oops, seems like you're unauthorised to do this. Try another API key with more rights or contact Nilvera.")
|
||
|
|
elif 403 < response.status_code < 600:
|
||
|
|
raise UserError("Odoo could not perform this action at the moment, try again later.\n%s - %s" % (response.reason, response.code))
|
||
|
|
|
||
|
|
try:
|
||
|
|
return response.json()
|
||
|
|
except JSONDecodeError:
|
||
|
|
_logger.exception("Invalid JSON response: %s", response.text)
|
||
|
|
raise UserError("An error occurred. Try again later.")
|