You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

126 lines
4.1 KiB

# Copyright 2020 Coop IT Easy SCRL fs
# Robin Keunen <robin@coopiteasy.be>
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
import json
import logging
import requests
from werkzeug.exceptions import BadRequest, InternalServerError, NotFound
from odoo import _, api, fields, models
from odoo.exceptions import AccessDenied, Warning as UserError
_logger = logging.getLogger(__name__)
class EMCBackend(models.Model):
_name = "emc.backend"
_description = "EMC Backend"
name = fields.Char(string="Name", required=True)
location = fields.Char(string="URL")
api_key = fields.Char(string="API Key")
description = fields.Text(string="Description", required=False)
active = fields.Boolean(string="active", default=True)
@api.model
def get_backend(self):
backend = self.env["emc.backend"].search([("active", "=", True)])
try:
backend.ensure_one()
except ValueError as e:
_logger.error(
"One and only one backend is allowed for the Easy My Coop "
"connector."
)
raise e
return backend
@api.multi
def http_get(self, url, params=None, headers=None):
self.ensure_one()
headers = self._add_api_key(headers)
if url.startswith("/"):
url = self.location + url
_logger.info("GET to {} w/ params {}".format(url, params))
return requests.get(url, params=params, headers=headers)
def _process_response(self, response):
if response.status_code == 200:
content = response.content.decode("utf-8")
return json.loads(content)
elif response.status_code == 400:
content = response.content.decode("utf-8")
raise BadRequest(
_(
"request returned status code %s with message %s"
% (response.status_code, content)
)
)
elif response.status_code == 403:
raise AccessDenied(
_("You are not allowed to access this resource")
)
elif response.status_code == 404:
raise NotFound(
_("Resource not found %s on server" % response.status_code)
)
else: # 500 et al.
content = response.content.decode("utf-8")
raise InternalServerError(
_(
"request returned status code %s with message %s"
% (response.status_code, content)
)
)
@api.multi
def http_get_content(self, url, params=None, headers=None):
self.ensure_one()
response = self.http_get(url, params=params, headers=headers)
return self._process_response(response)
@api.multi
def http_post(self, url, data, headers=None):
self.ensure_one()
headers = self._add_api_key(headers)
if url.startswith("/"):
url = self.location + url
_logger.info("POST to %s %s" % url)
return requests.post(url, json=data, headers=headers)
def http_post_content(self, url, data, headers=None):
self.ensure_one()
response = self.http_post(url, data, headers=headers)
return self._process_response(response)
@api.multi
def _add_api_key(self, headers):
self.ensure_one()
key_dict = {"API-KEY": self.api_key}
if headers:
headers.update(key_dict)
else:
headers = key_dict
return headers
@api.multi
def action_ping(self):
self.ensure_one()
url = self.location + "/api/ping/test"
try:
response = requests.get(url)
except Exception as e:
_logger.error(e)
raise UserError(_("Failed to connect to backend: %s" % str(e)))
if response.status_code == 200:
content = json.loads(response.content.decode("utf-8"))
raise UserError(_("Success: %s") % content["message"])
else:
raise UserError(
_("Failed to connect to backend: %s" % str(response.content))
)