Changeset - ab8894ef0db7
[Not reviewed]
0 4 1
Lance Edgar - 5 years ago 2020-03-15 14:28:22
lance@edbob.org
Add importer, datasync for CORE-POS (API) -> Rattail
5 files changed with 250 insertions and 35 deletions:
0 comments (0 inline, 0 general) First comment
rattail_corepos/commands.py
Show inline comments
 
@@ -45,6 +45,23 @@ class ExportCore(commands.ImportSubcommand):
 
        return load_object(spec)
 

	
 

	
 
class ImportCOREPOSAPI(commands.ImportSubcommand):
 
    """
 
    Import data from a CORE POS API
 
    """
 
    name = 'import-corepos-api'
 
    description = __doc__.strip()
 
    default_handler_spec = 'rattail_corepos.importing.corepos.api:FromCOREPOSToRattail'
 

	
 
    def get_handler_factory(self, **kwargs):
 
        if self.config:
 
            spec = self.config.get('rattail.importing', 'corepos_api.handler',
 
                                   default=self.default_handler_spec)
 
        else:
 
            spec = self.default_handler_spec
 
        return load_object(spec)
 

	
 

	
 
class ImportCOREPOSDB(commands.ImportSubcommand):
 
    """
 
    Import data from a CORE POS database
rattail_corepos/datasync/__init__.py
Show inline comments
 
# -*- coding: utf-8; -*-
 
################################################################################
 
#
 
#  Rattail -- Retail Software Framework
 
#  Copyright © 2010-2018 Lance Edgar
 
#
 
#  This file is part of Rattail.
 
#
 
#  Rattail is free software: you can redistribute it and/or modify it under the
 
#  terms of the GNU General Public License as published by the Free Software
 
#  Foundation, either version 3 of the License, or (at your option) any later
 
#  version.
 
#
 
#  Rattail is distributed in the hope that it will be useful, but WITHOUT ANY
 
#  WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 
#  FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
 
#  details.
 
#
 
#  You should have received a copy of the GNU General Public License along with
 
#  Rattail.  If not, see <http://www.gnu.org/licenses/>.
 
#
 
################################################################################
 
"""
 
CORE POS Interface
 
"""
 

	
 
from __future__ import unicode_literals, absolute_import
 

	
 
from rattail_corepos._version import __version__
rattail_corepos/datasync/rattail.py
Show inline comments
 
@@ -2,7 +2,7 @@
 
################################################################################
 
#
 
#  Rattail -- Retail Software Framework
 
#  Copyright © 2010-2018 Lance Edgar
 
#  Copyright © 2010-2020 Lance Edgar
 
#
 
#  This file is part of Rattail.
 
#
 
@@ -21,23 +21,47 @@
 
#
 
################################################################################
 
"""
 
DataSync for Milo
 
DataSync for Rattail DB
 
"""
 

	
 
from __future__ import unicode_literals, absolute_import
 

	
 
from sqlalchemy.orm.exc import NoResultFound
 

	
 
from corepos.api import CoreWebAPI
 
from corepos.db.office_op import Session as CoreSession, model as corepos
 

	
 
from rattail.datasync import NewDataSyncImportConsumer
 

	
 
from corepos.db import Session as CoreSession, model as corepos
 

	
 
class FromCOREAPIToRattail(NewDataSyncImportConsumer):
 
    """
 
    Consumer for CORE POS (API) -> Rattail datasync
 
    """
 
    handler_spec = 'rattail_corepos.importing.corepos.api:FromCOREPOSToRattail'
 

	
 
    def setup(self):
 
        super(FromCOREAPIToRattail, self).setup()
 
        self.establish_api()
 

	
 
    def establish_api(self):
 
        url = self.config.require('corepos.api', 'url')
 
        self.api = CoreWebAPI(url)
 

	
 
    def get_host_object(self, session, change):
 
        if change.payload_type == 'Department':
 
            return self.api.get_department(change.payload_key)
 
        if change.payload_type == 'Subdepartment':
 
            return self.api.get_subdepartment(change.payload_key)
 
        if change.payload_type == 'Vendor':
 
            return self.api.get_vendor(change.payload_key)
 
        if change.payload_type == 'Product':
 
            return self.api.get_product(change.payload_key)
 

	
 

	
 
class FromCOREPOSToRattailBase(NewDataSyncImportConsumer):
 
    """
 
    Base class for CORE POS -> Rattail data sync consumers.
 
    """
 
    handler_spec = 'rattail_corepos.importing.corepos:FromCOREPOSToRattail'
 
    handler_spec = 'rattail_corepos.importing.corepos.db:FromCOREPOSToRattail'
 

	
 
    def begin_transaction(self):
 
        self.corepos_session = CoreSession()
rattail_corepos/importing/corepos/api.py
Show inline comments
 
new file 100644
 
# -*- coding: utf-8; -*-
 
################################################################################
 
#
 
#  Rattail -- Retail Software Framework
 
#  Copyright © 2010-2020 Lance Edgar
 
#
 
#  This file is part of Rattail.
 
#
 
#  Rattail is free software: you can redistribute it and/or modify it under the
 
#  terms of the GNU General Public License as published by the Free Software
 
#  Foundation, either version 3 of the License, or (at your option) any later
 
#  version.
 
#
 
#  Rattail is distributed in the hope that it will be useful, but WITHOUT ANY
 
#  WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 
#  FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
 
#  details.
 
#
 
#  You should have received a copy of the GNU General Public License along with
 
#  Rattail.  If not, see <http://www.gnu.org/licenses/>.
 
#
 
################################################################################
 
"""
 
CORE POS (API) -> Rattail data importing
 
"""
 

	
 
import decimal
 
import logging
 

	
 
from corepos.api import CoreWebAPI
 

	
 
from rattail import importing
 
from rattail.gpc import GPC
 
from rattail.util import OrderedDict
 
from rattail_corepos import importing as corepos_importing
 

	
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
class FromCOREPOSToRattail(importing.ToRattailHandler):
 
    """
 
    Import handler for data coming from a CORE POS API.
 
    """
 
    host_title = "CORE-POS (API)"
 

	
 
    def get_importers(self):
 
        importers = OrderedDict()
 
        importers['Department'] = DepartmentImporter
 
        importers['Subdepartment'] = SubdepartmentImporter
 
        importers['Vendor'] = VendorImporter
 
        importers['Product'] = ProductImporter
 
        return importers
 

	
 

	
 
class FromCOREPOSAPI(importing.Importer):
 
    """
 
    Base class for all CORE POS API data importers.
 
    """
 

	
 
    def setup(self):
 
        super(FromCOREPOSAPI, self).setup()
 
        self.establish_api()
 

	
 
    def establish_api(self):
 
        url = self.config.require('corepos.api', 'url')
 
        self.api = CoreWebAPI(url)
 

	
 

	
 
class DepartmentImporter(FromCOREPOSAPI, importing.model.DepartmentImporter):
 
    """
 
    Importer for department data from CORE POS API.
 
    """
 
    key = 'number'
 
    supported_fields = [
 
        'number',
 
        'name',
 
    ]
 

	
 
    def get_host_objects(self):
 
        return self.api.get_departments()
 

	
 
    def normalize_host_object(self, department):
 
        return {
 
            'number': int(department['dept_no']),
 
            'name': department['dept_name'],
 
        }
 

	
 

	
 
class SubdepartmentImporter(FromCOREPOSAPI, importing.model.SubdepartmentImporter):
 
    """
 
    Importer for subdepartment data from CORE POS API.
 
    """
 
    key = 'number'
 
    supported_fields = [
 
        'number',
 
        'name',
 
        'department_number',
 
    ]
 

	
 
    def get_host_objects(self):
 
        return self.api.get_subdepartments()
 

	
 
    def normalize_host_object(self, subdepartment):
 
        return {
 
            'number': int(subdepartment['subdept_no']),
 
            'name': subdepartment['subdept_name'],
 
            'department_number': int(subdepartment['dept_ID']),
 
        }
 

	
 

	
 
class VendorImporter(FromCOREPOSAPI, corepos_importing.model.VendorImporter):
 
    """
 
    Importer for vendor data from CORE POS API.
 
    """
 
    key = 'corepos_id'
 
    supported_fields = [
 
        'corepos_id',
 
        'name',
 
        'abbreviation',
 
        'special_discount',
 
        'phone_number',
 
        'fax_number',
 
        'email_address',
 
    ]
 

	
 
    def get_host_objects(self):
 
        return self.api.get_vendors()
 

	
 
    def normalize_host_object(self, vendor):
 
        return {
 
            'corepos_id': int(vendor['vendorID']),
 
            'name': vendor['vendorName'],
 
            'abbreviation': vendor['vendorAbbreviation'] or None,
 
            'special_discount': decimal.Decimal(vendor['discountRate']),
 
            'phone_number': vendor.get('phone') or None,
 
            'fax_number': vendor.get('fax') or None,
 
            'email_address': vendor.get('email') or None,
 
        }
 

	
 

	
 
class ProductImporter(FromCOREPOSAPI, importing.model.ProductImporter):
 
    """
 
    Importer for product data from CORE POS API.
 
    """
 
    key = 'item_id'
 
    supported_fields = [
 
        'item_id',
 
        'upc',
 
        'brand_name',
 
        'description',
 
        'size',
 
        'weighed',
 
        'department_number',
 
        'subdepartment_number',
 
        'regular_price_price',
 
        'regular_price_multiple',
 
        'regular_price_type',
 
        'food_stampable',
 
        'tax1',
 
        'tax2',
 
    ]
 

	
 
    def get_host_objects(self):
 
        return self.api.get_products()
 

	
 
    def normalize_host_object(self, product):
 
        try:
 
            upc = GPC(product['upc'], calc_check_digit='upc')
 
        except (TypeError, ValueError):
 
            log.debug("CORE POS product has invalid UPC: %s", product['upc'])
 
            if len(self.key) == 1 and self.key[0] == 'upc':
 
                return
 
            upc = None
 

	
 
        price = None
 
        if product['normal_price'] is not None:
 
            price = decimal.Decimal(product['normal_price'])
 

	
 
        size = product.get('size', '').strip() or None
 
        if size == '0': # TODO: this is maybe just for sake of CORE sample data?
 
            size = None
 

	
 
        return {
 
            'item_id': product['upc'],
 
            'upc': upc,
 
            'brand_name': product.get('brand') or None,
 
            'description': product.get('description') or '',
 
            'size': size,
 

	
 
            'department_number': int(product['department']) or None,
 
            'subdepartment_number': int(product['subdept']) or None,
 

	
 
            'weighed': product['scale'] == '1',
 
            'food_stampable': product['foodstamp'] == '1',
 
            'tax1': product['tax'] == '1', # TODO: is this right?
 
            'tax2': product['tax'] == '2', # TODO: is this right?
 

	
 
            'regular_price_price': price,
 
            'regular_price_multiple': 1 if price is not None else None,
 
            'regular_price_type': self.enum.PRICE_TYPE_REGULAR if price is not None else None,
 
        }
setup.py
Show inline comments
 
@@ -114,6 +114,7 @@ setup(
 
        'rattail.commands': [
 
            'export-corepos = rattail_corepos.commands:ExportCore',
 
            'corepos-import-square = rattail_corepos.commands:CoreImportSquare',
 
            'import-corepos-api = rattail_corepos.commands:ImportCOREPOSAPI',
 
            'import-corepos-db = rattail_corepos.commands:ImportCOREPOSDB',
 
        ],
 

	
0 comments (0 inline, 0 general) First comment
You need to be logged in to comment. Login now