Source code for odoo.addons.connector_magento.components.backend_adapter

# -*- coding: utf-8 -*-
# Copyright 2017 Camptocamp SA
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html)

import socket
import logging
import xmlrpclib

from odoo.addons.component.core import AbstractComponent
from odoo.addons.queue_job.exception import RetryableJobError
from odoo.addons.connector.exception import NetworkRetryableError
from datetime import datetime

_logger = logging.getLogger(__name__)

try:
    import magento as magentolib
except ImportError:
    _logger.debug("Cannot import 'magento'")


MAGENTO_DATETIME_FORMAT = '%Y-%m-%d %H:%M:%S'


[docs]class MagentoLocation(object): def __init__(self, location, username, password, use_custom_api_path=False): self._location = location self.username = username self.password = password self.use_custom_api_path = use_custom_api_path self.use_auth_basic = False self.auth_basic_username = None self.auth_basic_password = None @property def location(self): location = self._location if not self.use_auth_basic: return location assert self.auth_basic_username and self.auth_basic_password replacement = "%s:%s@" % (self.auth_basic_username, self.auth_basic_password) location = location.replace('://', '://' + replacement) return location
[docs]class MagentoAPI(object): def __init__(self, location): """ :param location: Magento location :type location: :class:`MagentoLocation` """ self._location = location self._api = None @property def api(self): if self._api is None: custom_url = self._location.use_custom_api_path api = magentolib.API( self._location.location, self._location.username, self._location.password, full_url=custom_url ) api.__enter__() self._api = api return self._api def __enter__(self): # we do nothing, api is lazy return self def __exit__(self, type, value, traceback): if self._api is not None: self._api.__exit__(type, value, traceback)
[docs] def call(self, method, arguments): try: # When Magento is installed on PHP 5.4+, the API # may return garble data if the arguments contain # trailing None. if isinstance(arguments, list): while arguments and arguments[-1] is None: arguments.pop() start = datetime.now() try: result = self.api.call(method, arguments) except: _logger.error("api.call('%s', %s) failed", method, arguments) raise else: _logger.debug("api.call('%s', %s) returned %s in %s seconds", method, arguments, result, (datetime.now() - start).seconds) # Uncomment to record requests/responses in ``recorder`` # record(method, arguments, result) return result except (socket.gaierror, socket.error, socket.timeout) as err: raise NetworkRetryableError( 'A network error caused the failure of the job: ' '%s' % err) except xmlrpclib.ProtocolError as err: if err.errcode in [502, # Bad gateway 503, # Service unavailable 504]: # Gateway timeout raise RetryableJobError( 'A protocol error caused the failure of the job:\n' 'URL: %s\n' 'HTTP/HTTPS headers: %s\n' 'Error code: %d\n' 'Error message: %s\n' % (err.url, err.headers, err.errcode, err.errmsg)) else: raise
[docs]class MagentoCRUDAdapter(AbstractComponent): """ External Records Adapter for Magento """ _name = 'magento.crud.adapter' _inherit = ['base.backend.adapter', 'base.magento.connector'] _usage = 'backend.adapter'
[docs] def search(self, filters=None): """ Search records according to some criterias and returns a list of ids """ raise NotImplementedError
[docs] def read(self, id, attributes=None): """ Returns the information of a record """ raise NotImplementedError
[docs] def search_read(self, filters=None): """ Search records according to some criterias and returns their information""" raise NotImplementedError
[docs] def create(self, data): """ Create a record on the external system """ raise NotImplementedError
[docs] def write(self, id, data): """ Update records on the external system """ raise NotImplementedError
[docs] def delete(self, id): """ Delete a record on the external system """ raise NotImplementedError
[docs] def _call(self, method, arguments): try: magento_api = getattr(self.work, 'magento_api') except AttributeError: raise AttributeError( 'You must provide a magento_api attribute with a ' 'MagentoAPI instance to be able to use the ' 'Backend Adapter.' ) return magento_api.call(method, arguments)
[docs]class GenericAdapter(AbstractComponent): _name = 'magento.adapter' _inherit = 'magento.crud.adapter' _magento_model = None _admin_path = None
[docs] def search(self, filters=None): """ Search records according to some criterias and returns a list of ids :rtype: list """ return self._call('%s.search' % self._magento_model, [filters] if filters else [{}])
[docs] def read(self, id, attributes=None): """ Returns the information of a record :rtype: dict """ arguments = [int(id)] if attributes: # Avoid to pass Null values in attributes. Workaround for # https://bugs.launchpad.net/openerp-connector-magento/+bug/1210775 # When Magento is installed on PHP 5.4 and the compatibility patch # http://magento.com/blog/magento-news/magento-now-supports-php-54 # is not installed, calling info() with None in attributes # would return a wrong result (almost empty list of # attributes). The right correction is to install the # compatibility patch on Magento. arguments.append(attributes) return self._call('%s.info' % self._magento_model, arguments)
[docs] def search_read(self, filters=None): """ Search records according to some criterias and returns their information""" return self._call('%s.list' % self._magento_model, [filters])
[docs] def create(self, data): """ Create a record on the external system """ return self._call('%s.create' % self._magento_model, [data])
[docs] def write(self, id, data): """ Update records on the external system """ return self._call('%s.update' % self._magento_model, [int(id), data])
[docs] def delete(self, id): """ Delete a record on the external system """ return self._call('%s.delete' % self._magento_model, [int(id)])
[docs] def admin_url(self, id): """ Return the URL in the Magento admin for a record """ if self._admin_path is None: raise ValueError('No admin path is defined for this record') backend = self.backend_record url = backend.admin_location if not url: raise ValueError('No admin URL configured on the backend.') path = self._admin_path.format(model=self._magento_model, id=id) url = url.rstrip('/') path = path.lstrip('/') url = '/'.join((url, path)) return url