docs: update project links, kallithea -> forgejo

we were already "trying" to match on date only, but only as a sort of
fallback.  now we still try "exact" date/time match first but then
also an explicit date match, before other fallback logic

CHANGELOG.md

@@ -5,6 +5,64 @@ All notable changes to rattail-corepos will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
+## v0.3.9 (2024-08-19)
+
+### Fix
+
+- improve logic for matching CORE stock purchase to Rattail payment
+
+## v0.3.8 (2024-08-18)
+
+### Fix
+
+- avoid deprecated base class for config extension
+- work around, log error when datasync can't locate member
+
+## v0.3.7 (2024-08-13)
+
+### Fix
+
+- improve `core-office anonymize` command logic
+
+## v0.3.6 (2024-08-06)
+
+### Fix
+
+- fix DELETE triggers for `meminfo`, `employees`
+- avoid deprecated `AppProvider.load_object()` method
+
+## v0.3.5 (2024-07-14)
+
+### Fix
+
+- update app provider entry point, per wuttjamaican
+- fix CORE op model reference
+
+## v0.3.4 (2024-07-13)
+
+### Fix
+
+- refactor `config.get_model()` => `app.model`
+- avoid error when CORE API gives record with no upc
+
+## v0.3.3 (2024-07-05)
+
+### Fix
+
+- add logic to auto-create user for CORE POS cashier login
+- fix employee status when importing from CORE API
+- add Employee support for CORE API -> Rattail import/datasync
+- misc. improvements for CORE API importer, per flaky data
+- add command to install mysql triggers for CORE `office_op` DB
+- improve ProductCost sorting for import from CORE API
+- include `person_uuid` for Member import from CORE API
+
+## v0.3.2 (2024-07-02)
+
+### Fix
+
+- avoid deprecated function for `get_engines()` config
+
 ## v0.3.1 (2024-07-01)
 
 ### Fix

README.md

@@ -0,0 +1,11 @@
+
+# rattail-corepos
+
+Rattail is a retail software framework, released under the GNU General Public
+License.
+
+This package contains software interfaces for the [CORE
+POS](https://github.com/CORE-POS/IS4C) system.
+
+Please see Rattail's [home page](https://rattailproject.org/) for more
+information.

README.rst

@@ -1,14 +0,0 @@
-
-rattail_corepos
-===============
-
-Rattail is a retail software framework, released under the GNU General Public
-License.
-
-This package contains software interfaces for the `CORE POS`_ system.
-
-.. _`CORE POS`: https://github.com/CORE-POS/IS4C
-
-Please see Rattail's `home page`_ for more information.
-
-.. _`home page`: https://rattailproject.org/

pyproject.toml

@@ -6,9 +6,9 @@ build-backend = "hatchling.build"
 
 [project]
 name = "rattail_corepos"
-version = "0.3.1"
+version = "0.3.9"
 description = "Rattail Software Interfaces for CORE POS"
-readme = "README.rst"
+readme = "README.md"
 authors = [{name = "Lance Edgar", email = "lance@edbob.org"}]
 license = {text = "GNU GPL v3+"}
 classifiers = [
@@ -33,10 +33,10 @@ dependencies = [
 
 
 [project.urls]
-Homepage = "https://redmine.rattailproject.org/projects/corepos-integration"
-Repository = "https://kallithea.rattailproject.org/rattail-project/rattail-corepos"
-Issues = "https://redmine.rattailproject.org/projects/corepos-integration/issues"
-Changelog = "https://kallithea.rattailproject.org/rattail-project/rattail-corepos/files/master/CHANGELOG.md"
+Homepage = "https://rattailproject.org"
+Repository = "https://forgejo.wuttaproject.org/rattail/rattail-corepos"
+Issues = "https://forgejo.wuttaproject.org/rattail/rattail-corepos/issues"
+Changelog = "https://forgejo.wuttaproject.org/rattail/rattail-corepos/src/branch/master/CHANGELOG.md"
 
 
 [project.scripts]
@@ -64,7 +64,7 @@ rattail_corepos = "rattail_corepos.emails"
 "to_trainwreck.from_corepos_db_office_trans" = "rattail_corepos.trainwreck.importing.corepos:FromCoreToTrainwreck"
 
 
-[project.entry-points."rattail.providers"]
+[project.entry-points."wutta.app.providers"]
 rattail_corepos = "rattail_corepos.app:CoreProvider"
 
 

rattail_corepos/app.py

@@ -34,9 +34,9 @@ class CoreProvider(RattailProvider):
 
     def get_corepos_handler(self, **kwargs):
         if not hasattr(self, 'corepos_handler'):
-            spec = self.config.get('rattail', 'corepos.handler',
+            spec = self.config.get('rattail.corepos.handler',
                                    default='rattail_corepos.app:CoreHandler')
-            factory = self.load_object(spec)
+            factory = self.app.load_object(spec)
             self.corepos_handler = factory(self.config, **kwargs)
         return self.corepos_handler
 

rattail_corepos/auth.py

@@ -2,7 +2,7 @@
 ################################################################################
 #
 #  Rattail -- Retail Software Framework
-#  Copyright © 2010-2023 Lance Edgar
+#  Copyright © 2010-2024 Lance Edgar
 #
 #  This file is part of Rattail.
 #
@@ -50,6 +50,15 @@ class CoreAuthHandler(base.AuthHandler):
             core_employee = self.check_corepos_cashier_credentials(core_session, password)
             if core_employee:
                 user = self.get_user_from_corepos_employee(session, core_employee)
+                if not user and self.config.get_bool('rattail.auth.corepos.automake_users'):
+                    # nb. new user must be made via separate session
+                    # and then merged back into the main session.
+                    # this is because the caller cannot be responsible
+                    # for committing (persisting) the new user.
+                    with self.app.short_session() as s:
+                        user = self.make_user_from_corepos_employee(s, core_employee)
+                        s.commit()
+                        user = session.get(model.User, user.uuid)
             core_session.close()
             if user and user.active:
                 return user
@@ -67,14 +76,33 @@ class CoreAuthHandler(base.AuthHandler):
             if core_employee.active:
                 return core_employee
 
-    def get_user_from_corepos_employee(self, session, core_employee):
+    def get_rattail_employee(self, session, core_employee):
         model = self.model
         try:
-            employee = session.query(model.Employee)\
-                              .join(model.CoreEmployee)\
-                              .filter(model.CoreEmployee.corepos_number == core_employee.number)\
-                              .one()
+            return session.query(model.Employee)\
+                          .join(model.CoreEmployee)\
+                          .filter(model.CoreEmployee.corepos_number == core_employee.number)\
+                          .one()
         except orm.exc.NoResultFound:
             pass
-        else:
+
+    def get_user_from_corepos_employee(self, session, core_employee):
+        employee = self.get_rattail_employee(session, core_employee)
+        if employee:
             return self.app.get_user(employee)
+
+    def make_user_from_corepos_employee(self, session, core_employee):
+        employee = self.get_rattail_employee(session, core_employee)
+        if not employee:
+            raise ValueError(f"CORE employee not found in {self.app.get_title()}: {core_employee}")
+
+        person = self.app.get_person(employee)
+        user = self.make_user(session=session, person=person)
+
+        role = self.config.get('rattail.auth.corepos.automake_users_role')
+        if role:
+            role = self.get_role(session, role)
+            if role:
+                user.roles.append(role)
+
+        return user

rattail_corepos/config.py

@@ -2,7 +2,7 @@
 ################################################################################
 #
 #  Rattail -- Retail Software Framework
-#  Copyright © 2010-2023 Lance Edgar
+#  Copyright © 2010-2024 Lance Edgar
 #
 #  This file is part of Rattail.
 #
@@ -26,11 +26,11 @@ Rattail-COREPOS Config Extension
 
 import warnings
 
-from rattail.config import ConfigExtension
-from rattail.db.config import get_engines
+from wuttjamaican.conf import WuttaConfigExtension
+from wuttjamaican.db.conf import get_engines
 
 
-class RattailCOREPOSExtension(ConfigExtension):
+class RattailCOREPOSExtension(WuttaConfigExtension):
     """
     Config extension for Rattail-COREPOS
     """
@@ -45,7 +45,7 @@ class RattailCOREPOSExtension(ConfigExtension):
 
         # office_op
         from corepos.db.office_op import Session
-        engines = get_engines(config, section='corepos.db.office_op')
+        engines = get_engines(config, 'corepos.db.office_op')
         config.core_office_op_engines = engines
         config.core_office_op_engine = engines.get('default')
         Session.configure(bind=config.core_office_op_engine)
@@ -55,7 +55,7 @@ class RattailCOREPOSExtension(ConfigExtension):
 
         # office_trans
         from corepos.db.office_trans import Session
-        engines = get_engines(config, section='corepos.db.office_trans')
+        engines = get_engines(config, 'corepos.db.office_trans')
         config.core_office_trans_engines = engines
         config.core_office_trans_engine = engines.get('default')
         Session.configure(bind=config.core_office_trans_engine)
@@ -65,9 +65,9 @@ class RattailCOREPOSExtension(ConfigExtension):
 
         # office_arch
         from corepos.db.office_arch import Session
-        engines = get_engines(config, section='corepos.db.office_arch')
+        engines = get_engines(config, 'corepos.db.office_arch')
         if not engines:
-            engines = get_engines(config, section='corepos.db.office_trans_archive')
+            engines = get_engines(config, 'corepos.db.office_trans_archive')
             if engines:
                 warnings.warn("config section [corepos.db.office_trans_archive] is deprecated; "
                               "please use section [corepos.db.office_arch] instead",
@@ -81,7 +81,7 @@ class RattailCOREPOSExtension(ConfigExtension):
 
         # lane_op
         from corepos.db.lane_op import Session
-        engines = get_engines(config, section='corepos.db.lane_op')
+        engines = get_engines(config, 'corepos.db.lane_op')
         config.core_lane_op_engines = engines
         config.core_lane_op_engine = engines.get('default')
         Session.configure(bind=config.core_lane_op_engine)

rattail_corepos/corepos/office/anonymize.py

@@ -37,8 +37,13 @@ class Anonymizer(GenericHandler):
     """
 
     def anonymize_all(self, dbkey=None, dry_run=False, progress=None):
+        # nb. these must be installed with:
+        #   pip install names us zipcodes
         import names
         import us
+        import zipcodes
+
+        self.all_zipcodes = zipcodes.list_all()
 
         core_handler = self.app.get_corepos_handler()
         op_session = core_handler.make_session_office_op(dbkey=dbkey)
@@ -46,45 +51,56 @@ class Anonymizer(GenericHandler):
 
         states = [state.abbr for state in us.states.STATES]
 
-        # meminfo
-        members = op_session.query(op_model.MemberInfo).all()
-        members_by_card_number = {}
-
-        def anon_meminfo(member, i):
-            member.first_name = names.get_first_name()
-            member.last_name = names.get_last_name()
-            member.other_first_name = names.get_first_name()
-            member.other_last_name = names.get_last_name()
-            member.street = '123 Main St.'
-            member.city = 'Anytown'
-            member.state = random.choice(states)
-            member.zipcode = self.random_zipcode()
-            member.phone = self.random_phone()
-            member.email = self.random_email()
-            member.notes.clear()
-            members_by_card_number[member.card_number] = member
-
-        self.app.progress_loop(anon_meminfo, members, progress,
-                               message="Anonymizing meminfo")
-
         # custdata
-        customers = op_session.query(op_model.CustomerClassic).all()
+        customers = op_session.query(op_model.CustomerClassic)\
+                              .order_by(op_model.CustomerClassic.card_number,
+                                        op_model.CustomerClassic.person_number)\
+                              .all()
         blueline_template = get_blueline_template(self.config)
+        customers_by_card_number = {}
 
         def anon_custdata(customer, i):
-            member = members_by_card_number.get(customer.card_number)
-            if member:
-                customer.first_name = member.first_name
-                customer.last_name = member.last_name
-            else:
-                customer.first_name = names.get_first_name()
-                customer.last_name = names.get_last_name()
+            customer.first_name = names.get_first_name()
+            customer.last_name = names.get_last_name()
             customer.blue_line = make_blueline(self.config, customer,
                                                template=blueline_template)
+            customers_by_card_number.setdefault(customer.card_number, []).append(customer)
 
         self.app.progress_loop(anon_custdata, customers, progress,
                                message="Anonymizing custdata")
 
+        # meminfo
+        members = op_session.query(op_model.MemberInfo).all()
+
+        def anon_meminfo(member, i):
+            if member.first_name:
+                member.first_name = names.get_first_name()
+            if member.last_name:
+                member.last_name = names.get_last_name()
+            if member.other_first_name:
+                member.other_first_name = names.get_first_name()
+            if member.other_last_name:
+                member.other_last_name = names.get_last_name()
+
+            member.street = '123 Main St.'
+            member.city = 'Anytown'
+            member.state = random.choice(states)
+            member.zip = self.random_zipcode()
+            member.phone = self.random_phone()
+
+            customers = customers_by_card_number.get(member.card_number)
+            if customers:
+                customer = customers[0]
+                member.email = f'{customer.first_name}_{customer.last_name}@mailinator.com'\
+                      .lower()
+            else:
+                member.email = self.random_email()
+
+            member.notes.clear()
+
+        self.app.progress_loop(anon_meminfo, members, progress,
+                               message="Anonymizing meminfo")
+
         # Customers
         customers = op_session.query(op_model.Customer).all()
 
@@ -133,9 +149,7 @@ class Anonymizer(GenericHandler):
         import names
         name = names.get_full_name()
         name = name.replace(' ', '_')
-        return f'{name}@mailinator.com'
+        return f'{name}@mailinator.com'.lower()
 
     def random_zipcode(self):
-        digits = [random.choice('0123456789')
-                  for i in range(5)]
-        return ''.join(digits)
+        return random.choice(self.all_zipcodes)['zip_code']

rattail_corepos/corepos/office/commands.py

@@ -95,6 +95,13 @@ def anonymize(
                          "\tpip install us\n")
         sys.exit(2)
 
+    try:
+        import zipcodes
+    except ImportError:
+        sys.stderr.write("must install the `zipcodes` package first!\n\n"
+                         "\tpip install zipcodes\n")
+        sys.exit(2)
+
     anonymizer = Anonymizer(config)
     anonymizer.anonymize_all(dbkey=dbkey, dry_run=dry_run,
                              progress=progress)
@@ -183,6 +190,47 @@ def import_self(
     handler.run(kwargs, progress=progress)
 
 
+@core_office_typer.command()
+def install_triggers(
+        ctx: typer.Context,
+        status: Annotated[
+            bool,
+            typer.Option('--status',
+                         help="Show current status of DB, then exit.")] = False,
+        uninstall: Annotated[
+            bool,
+            typer.Option('--uninstall',
+                         help="Uninstall table and triggers, instead of install.")] = False,
+        table_name: Annotated[
+            str,
+            typer.Option(help="Override name of \"changes\" table if needed.")] = 'datasync_changes',
+        dry_run: Annotated[
+            bool,
+            typer.Option('--dry-run',
+                         help="Do not (un)install anything, but show what would have been done.")] = False,
+):
+    """
+    Install MySQL DB triggers for use with Rattail DataSync
+    """
+    from rattail_corepos.corepos.office.triggers import CoreTriggerHandler
+
+    config = ctx.parent.rattail_config
+    app = config.get_app()
+    corepos = app.get_corepos_handler()
+    op_session = corepos.make_session_office_op()
+    triggers = CoreTriggerHandler(config)
+
+    if status:
+        triggers.show_status(op_session, table_name)
+    elif uninstall:
+        triggers.uninstall_all(op_session, table_name, dry_run=dry_run)
+    else:
+        triggers.install_all(op_session, table_name, dry_run=dry_run)
+
+    op_session.commit()
+    op_session.close()
+
+
 @core_office_typer.command()
 def patch_customer_gaps(
         ctx: typer.Context,

rattail_corepos/corepos/office/triggers.py

@@ -0,0 +1,360 @@
+# -*- coding: utf-8; -*-
+################################################################################
+#
+#  Rattail -- Retail Software Framework
+#  Copyright © 2010-2024 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 Office - datasync triggers
+"""
+
+import sqlalchemy as sa
+
+from rattail.app import GenericHandler
+from rattail_corepos.datasync.corepos import make_changes_table
+
+
+class CoreTriggerHandler(GenericHandler):
+    """
+    Handler to install and show status of CORE DB triggers, for use
+    with Rattail DataSync.
+    """
+    supported_triggers = [
+        'custdata',
+        'meminfo',
+        'employees',
+        'departments',
+        'subdepts',
+        'vendors',
+        'products',
+        'vendorItems',
+    ]
+
+    def show_status(self, op_session, table_name):
+        """
+        Show trigger status for an ``office_op`` database.
+        """
+        print()
+        print("database")
+        # nb. use repr() to hide password
+        print(f"url: {repr(op_session.bind.url)}")
+        exists = self.database_exists(op_session)
+        print(f"exists: {exists}")
+        if not exists:
+            return # nothing more to test
+        print()
+
+        print("changes table")
+        print(f"name: {table_name}")
+        table = self.make_changes_table(table_name)
+        exists = self.changes_table_exists(op_session, table)
+        print(f"exists: {exists}")
+        if exists:
+            records = op_session.execute(table.select())
+            print(f"records: {len(records.fetchall())}")
+        print()
+
+        for trigger in self.supported_triggers:
+            print(f"triggers for {trigger}")
+            
+            create = f'record_{trigger}_create'
+            exists = self.trigger_exists(op_session, create)
+            print(f"{create:40s} exists: {exists}")
+
+            update = f'record_{trigger}_update'
+            exists = self.trigger_exists(op_session, update)
+            print(f"{update:40s} exists: {exists}")
+
+            delete = f'record_{trigger}_delete'
+            exists = self.trigger_exists(op_session, delete)
+            print(f"{delete:40s} exists: {exists}")
+
+            print()
+
+    def database_exists(self, op_session):
+        corepos = self.app.get_corepos_handler()
+        op_model = corepos.get_model_office_op()
+        try:
+            # just query a basic table, if things are normal then we're good
+            op_session.query(op_model.Department).count()
+        except sa.exc.ProgrammingError:
+            return False
+        return True
+
+    def trigger_exists(self, op_session, trigger):
+        dbname = op_session.bind.url.database
+        sql = sa.text(f"""
+        SHOW TRIGGERS FROM `{dbname}` WHERE `Trigger` = :trigger
+        """)
+        result = op_session.execute(sql, {'trigger': trigger})
+        if result.fetchone():
+            return True
+        return False
+
+    def changes_table_exists(self, op_session, table):
+        if isinstance(table, str):
+            table = self.make_changes_table(table)
+        try:
+            op_session.execute(table.select())
+        except sa.exc.ProgrammingError:
+            return False
+        return True
+
+    def make_changes_table(self, table_name):
+        metadata = sa.MetaData()
+        table = make_changes_table(table_name, metadata)
+        return table
+
+    def install_all(self, op_session, table_name, dry_run=False):
+        self.install_changes_table(op_session, table_name, dry_run=dry_run)
+        self.install_triggers(op_session, table_name, dry_run=dry_run)
+
+    def install_changes_table(self, op_session, table_name, dry_run=False):
+        print()
+        print("installing changes table...")
+        print(f"{table_name}: ", end='')
+
+        table = self.make_changes_table(table_name)
+        if self.changes_table_exists(op_session, table):
+            print("already exists")
+            print()
+            return
+
+        if not dry_run:
+            table.create(op_session.bind)
+        print("done")
+        print()
+
+    def install_triggers(self, op_session, table_name, dry_run=False):
+        print("installing triggers...")
+
+        for trigger in self.supported_triggers:
+            if not dry_run:
+                self.drop_triggers(op_session, trigger)
+
+                meth = getattr(self, f'create_triggers_{trigger}')
+                meth(op_session, table_name)
+
+        print("done")
+        print()
+
+    def uninstall_all(self, op_session, table_name, dry_run=False):
+        self.uninstall_changes_table(op_session, table_name, dry_run=dry_run)
+        self.uninstall_triggers(op_session, dry_run=dry_run)
+
+    def uninstall_changes_table(self, op_session, table_name, dry_run=False):
+        print()
+        print("uninstalling changes table...")
+
+        table = self.make_changes_table(table_name)
+        if not self.changes_table_exists(op_session, table):
+            print("table does not exist")
+            print()
+            return
+
+        if not dry_run:
+            # TODO: why does this drop() method just hang forever?
+            #table.drop(op_session.bind)
+            op_session.execute(sa.text(f"DROP TABLE {table_name}"))
+        print("done")
+        print()
+
+    def uninstall_triggers(self, op_session, dry_run=False):
+        print("uninstalling triggers...")
+
+        for trigger in self.supported_triggers:
+            if not dry_run:
+                self.drop_triggers(op_session, trigger)
+
+        print("done")
+        print()
+
+    def create_triggers_custdata(self, op_session, changes_table):
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_custdata_create
+        AFTER INSERT ON custdata
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Member', CONVERT(NEW.CardNo, CHAR), 0);
+        """))
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_custdata_update
+        AFTER UPDATE ON custdata
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Member', CONVERT(NEW.CardNo, CHAR), 0);
+        """))
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_custdata_delete
+        AFTER DELETE ON custdata
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Member', CONVERT(OLD.CardNo, CHAR), 1);
+        """))
+
+    def create_triggers_meminfo(self, op_session, changes_table):
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_meminfo_create
+        AFTER INSERT ON meminfo
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Member', CONVERT(NEW.card_no, CHAR), 0);
+        """))
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_meminfo_update
+        AFTER UPDATE ON meminfo
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Member', CONVERT(NEW.card_no, CHAR), 0);
+        """))
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_meminfo_delete
+        AFTER DELETE ON meminfo
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Member', CONVERT(OLD.card_no, CHAR), 1);
+        """))
+
+    def create_triggers_employees(self, op_session, changes_table):
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_employees_create
+        AFTER INSERT ON employees
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Employee', CONVERT(NEW.emp_no, CHAR), 0);
+        """))
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_employees_update
+        AFTER UPDATE ON employees
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Employee', CONVERT(NEW.emp_no, CHAR), 0);
+        """))
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_employees_delete
+        AFTER DELETE ON employees
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Employee', CONVERT(OLD.emp_no, CHAR), 1);
+        """))
+
+    def create_triggers_departments(self, op_session, changes_table):
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_departments_create
+        AFTER INSERT ON departments
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Department', CONVERT(NEW.dept_no, CHAR), 0);
+        """))
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_departments_update
+        AFTER UPDATE ON departments
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Department', CONVERT(NEW.dept_no, CHAR), 0);
+        """))
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_departments_delete
+        AFTER DELETE ON departments
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Department', CONVERT(OLD.dept_no, CHAR), 1);
+        """))
+
+    def create_triggers_subdepts(self, op_session, changes_table):
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_subdepts_create
+        AFTER INSERT ON subdepts
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Subdepartment', CONVERT(NEW.subdept_no, CHAR), 0);
+        """))
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_subdepts_update
+        AFTER UPDATE ON subdepts
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Subdepartment', CONVERT(NEW.subdept_no, CHAR), 0);
+        """))
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_subdepts_delete
+        AFTER DELETE ON subdepts
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Subdepartment', CONVERT(OLD.subdept_no, CHAR), 1);
+        """))
+
+    def create_triggers_vendors(self, op_session, changes_table):
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_vendors_create
+        AFTER INSERT ON vendors
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Vendor', CONVERT(NEW.vendorID, CHAR), 0);
+        """))
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_vendors_update
+        AFTER UPDATE ON vendors
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Vendor', CONVERT(NEW.vendorID, CHAR), 0);
+        """))
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_vendors_delete
+        AFTER DELETE ON vendors
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Vendor', CONVERT(OLD.vendorID, CHAR), 1);
+        """))
+
+    def create_triggers_products(self, op_session, changes_table):
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_products_create
+        AFTER INSERT ON products
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Product', NEW.upc, 0);
+        """))
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_products_update
+        AFTER UPDATE ON products
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Product', NEW.upc, 0);
+        """))
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_products_delete
+        AFTER DELETE ON products
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('Product', OLD.upc, 1);
+        """))
+
+    def create_triggers_vendorItems(self, op_session, changes_table):
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_vendorItems_create
+        AFTER INSERT ON vendorItems
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('VendorItem', CONCAT_WS('|', NEW.sku, CONVERT(NEW.vendorID, CHAR)), 0);
+        """))
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_vendorItems_update
+        AFTER UPDATE ON vendorItems
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('VendorItem', CONCAT_WS('|', NEW.sku, CONVERT(NEW.vendorID, CHAR)), 0);
+        """))
+
+        op_session.execute(sa.text(f"""
+        CREATE TRIGGER record_vendorItems_delete
+        AFTER DELETE ON vendorItems
+        FOR EACH ROW INSERT INTO {changes_table} (object_type, object_key, deleted) VALUES ('VendorItem', CONCAT_WS('|', OLD.sku, CONVERT(OLD.vendorID, CHAR)), 1);
+        """))
+
+    def drop_triggers(self, op_session, trigger):
+
+        op_session.execute(sa.text(f"""
+        DROP TRIGGER IF EXISTS record_{trigger}_create;
+        """))
+
+        op_session.execute(sa.text(f"""
+        DROP TRIGGER IF EXISTS record_{trigger}_update;
+        """))
+
+        op_session.execute(sa.text(f"""
+        DROP TRIGGER IF EXISTS record_{trigger}_delete;
+        """))

rattail_corepos/datasync/corepos.py

@@ -2,7 +2,7 @@
 ################################################################################
 #
 #  Rattail -- Retail Software Framework
-#  Copyright © 2010-2023 Lance Edgar
+#  Copyright © 2010-2024 Lance Edgar
 #
 #  This file is part of Rattail.
 #
@@ -26,12 +26,19 @@ DataSync for CORE POS
 
 import sqlalchemy as sa
 
-from corepos.db.office_op import Session as CoreSession, model as corepos
-
-from rattail.db import model
 from rattail.datasync import DataSyncWatcher, DataSyncImportConsumer
 
 
+def make_changes_table(table_name, metadata):
+    return sa.Table(
+        table_name, metadata,
+        sa.Column('id', sa.Integer(), nullable=False, primary_key=True),
+        sa.Column('object_type', sa.String(length=255), nullable=False),
+        sa.Column('object_key', sa.String(length=255), nullable=False),
+        sa.Column('deleted', sa.Boolean(), nullable=False, default=False),
+    )
+
+
 class CoreOfficeOpWatcher(DataSyncWatcher):
     """
     DataSync watcher for the CORE ``office_op`` database.
@@ -39,21 +46,19 @@ class CoreOfficeOpWatcher(DataSyncWatcher):
     prunes_changes = True
 
     def __init__(self, *args, **kwargs):
-        super(CoreOfficeOpWatcher, self).__init__(*args, **kwargs)
+        super().__init__(*args, **kwargs)
 
         self.changes_table_name = kwargs.get('changes_table_name',
                                              'datasync_changes')
 
         self.corepos_metadata = sa.MetaData()
-        self.corepos_changes = sa.Table(
-            self.changes_table_name, self.corepos_metadata,
-            sa.Column('id', sa.Integer(), nullable=False, primary_key=True),
-            sa.Column('object_type', sa.String(length=255), nullable=False),
-            sa.Column('object_key', sa.String(length=255), nullable=False),
-            sa.Column('deleted', sa.Boolean(), nullable=False, default=False))
+        self.corepos_changes = make_changes_table(self.changes_table_name,
+                                                  self.corepos_metadata)
 
     def get_changes(self, lastrun):
-        session = CoreSession()
+        model = self.model
+        corepos = self.app.get_corepos_handler()
+        session = corepos.make_session_office_op()
         result = session.execute(self.corepos_changes.select())
         changes = result.fetchall()
         session.close()
@@ -67,8 +72,9 @@ class CoreOfficeOpWatcher(DataSyncWatcher):
                 for c in changes]
 
     def prune_changes(self, keys):
+        corepos = self.app.get_corepos_handler()
+        session = corepos.make_session_office_op()
         deleted = 0
-        session = CoreSession()
         for key in keys:
             result = session.execute(self.corepos_changes.select()\
                                      .where(self.corepos_changes.c.id == key))
@@ -90,13 +96,17 @@ class COREPOSProductWatcher(DataSyncWatcher):
         if not lastrun:
             return
 
+        model = self.model
+        corepos = self.app.get_corepos_handler()
+        op_model = corepos.get_model_office_op()
+
         changes = []
-        session = CoreSession()
+        session = corepos.make_session_office_op()
         lastrun = self.localize_lastrun(session, lastrun)
 
         # Department
-        departments = session.query(corepos.Department)\
-                             .filter(corepos.Department.modified >= lastrun)\
+        departments = session.query(op_model.Department)\
+                             .filter(op_model.Department.modified >= lastrun)\
                              .all()
         if departments:
             changes.extend([
@@ -133,8 +143,8 @@ class COREPOSProductWatcher(DataSyncWatcher):
         #         for vendor in vendors])
 
         # Product
-        products = session.query(corepos.Product)\
-                          .filter(corepos.Product.modified >= lastrun)\
+        products = session.query(op_model.Product)\
+                          .filter(op_model.Product.modified >= lastrun)\
                           .all()
         if products:
             changes.extend([
@@ -236,9 +246,11 @@ class FromRattailToCore(DataSyncImportConsumer):
             self.invoke_importer(session, change)
 
     def get_host_object(self, session, change):
+        model = self.model
         return session.get(getattr(model, change.payload_type), change.payload_key)
 
     def get_customers(self, session, change):
+        model = self.model
         clientele = self.app.get_clientele_handler()
 
         if change.payload_type == 'Customer':
@@ -284,6 +296,7 @@ class FromRattailToCore(DataSyncImportConsumer):
         return []
 
     def get_vendor(self, session, change):
+        model = self.model
 
         if change.payload_type == 'Vendor':
             return session.get(model.Vendor, change.payload_key)
@@ -299,6 +312,7 @@ class FromRattailToCore(DataSyncImportConsumer):
                 return email.vendor
 
     def get_product(self, session, change):
+        model = self.model
 
         if change.payload_type == 'Product':
             return session.get(model.Product, change.payload_key)

rattail_corepos/datasync/rattail.py

@@ -2,7 +2,7 @@
 ################################################################################
 #
 #  Rattail -- Retail Software Framework
-#  Copyright © 2010-2023 Lance Edgar
+#  Copyright © 2010-2024 Lance Edgar
 #
 #  This file is part of Rattail.
 #
@@ -24,13 +24,16 @@
 DataSync for Rattail DB
 """
 
+import logging
+
 from sqlalchemy import orm
 
-from corepos.db.office_op import Session as CoreSession, model as corepos
-
 from rattail.datasync import DataSyncImportConsumer
 
 
+log = logging.getLogger(__name__)
+
+
 class FromCOREAPIToRattail(DataSyncImportConsumer):
     """
     Consumer for CORE POS (API) -> Rattail datasync
@@ -71,6 +74,11 @@ class FromCOREAPIToRattail(DataSyncImportConsumer):
             else:
                 # import member data from API, into various Rattail tables
                 member = self.get_host_object(session, change)
+                if not member:
+                    # TODO: should log.warning() instead but for now i
+                    # need to see this in action and further troubleshoot
+                    log.error("CORE member not found for change: %s", change)
+                    continue
                 self.process_change(session, self.importers['Customer'],
                                     host_object=member)
                 shoppers = self.importers['CustomerShopper'].get_shoppers_for_member(member)
@@ -80,6 +88,14 @@ class FromCOREAPIToRattail(DataSyncImportConsumer):
                 self.process_change(session, self.importers['Member'],
                                     host_object=member)
 
+        # sync all Employee-related changes
+        types = [
+            'Employee',
+        ]
+        for change in [c for c in changes if c.payload_type in types]:
+            # normal logic works fine here
+            self.invoke_importer(session, change)
+
         # sync all "product meta" changes
         types = [
             'Department',
@@ -120,6 +136,8 @@ class FromCOREAPIToRattail(DataSyncImportConsumer):
     def get_host_object(self, session, change):
         if change.payload_type == 'Member':
             return self.api.get_member(change.payload_key)
+        if change.payload_type == 'Employee':
+            return self.api.get_employee(change.payload_key)
         if change.payload_type == 'Department':
             return self.api.get_department(change.payload_key)
         if change.payload_type == 'Subdepartment':
@@ -143,7 +161,7 @@ class FromCOREAPIToRattail(DataSyncImportConsumer):
             if len(fields) == 2:
                 sku, vendorID = fields
                 vendor_item = self.api.get_vendor_item(sku, vendorID)
-                if vendor_item:
+                if vendor_item and vendor_item.get('upc'):
                     return self.api.get_product(vendor_item['upc'])
 
 
@@ -154,7 +172,8 @@ class FromCOREPOSToRattailBase(DataSyncImportConsumer):
     handler_spec = 'rattail_corepos.importing.corepos.db:FromCOREPOSToRattail'
 
     def begin_transaction(self):
-        self.corepos_session = CoreSession()
+        corepos = self.app.get_corepos_handler()
+        self.corepos_session = corepos.make_session_office_op()
 
     def rollback_transaction(self):
         self.corepos_session.rollback()
@@ -172,16 +191,18 @@ class FromCOREPOSToRattailProducts(FromCOREPOSToRattailBase):
     """
 
     def get_host_object(self, session, change):
+        corepos = self.app.get_corepos_handler()
+        op_model = corepos.get_model_office_op()
 
         if change.payload_type == 'Product':
             try:
-                return self.corepos_session.query(corepos.Product)\
-                                           .filter(corepos.Product.upc == change.payload_key)\
+                return self.corepos_session.query(op_model.Product)\
+                                           .filter(op_model.Product.upc == change.payload_key)\
                                            .one()
             except orm.exc.NoResultFound:
                 pass
 
         else:
             # try to fetch CORE POS object via typical method
-            Model = getattr(corepos, change.payload_type)
+            Model = getattr(op_model, change.payload_type)
             return self.corepos_session.get(Model, int(change.payload_key))

rattail_corepos/importing/corepos/api.py

@@ -29,6 +29,7 @@ import decimal
 import logging
 from collections import OrderedDict
 
+import sqlalchemy as sa
 from sqlalchemy import orm
 
 from rattail import importing
@@ -55,6 +56,7 @@ class FromCOREPOSToRattail(importing.ToRattailHandler):
         importers['CustomerShopper'] = CustomerShopperImporter
         importers['MembershipType'] = MembershipTypeImporter
         importers['Member'] = MemberImporter
+        importers['Employee'] = EmployeeImporter
         importers['Store'] = StoreImporter
         importers['Department'] = DepartmentImporter
         importers['Subdepartment'] = SubdepartmentImporter
@@ -340,6 +342,34 @@ class CustomerShopperImporter(FromCOREPOSAPI, corepos_importing.model.CustomerSh
         return data
 
 
+class EmployeeImporter(FromCOREPOSAPI, corepos_importing.model.EmployeeImporter):
+    """
+    Importer for employee data from CORE POS API.
+    """
+    key = 'corepos_number'
+    supported_fields = [
+        'corepos_number',
+        'id',
+        'first_name',
+        'last_name',
+        'full_name',
+        'status',
+    ]
+
+    def get_host_objects(self):
+        return self.api.get_employees()
+
+    def normalize_host_object(self, employee):
+        return {
+            'corepos_number': int(employee['emp_no']),
+            'id': int(employee['emp_no']),
+            'first_name': employee['FirstName'],
+            'last_name': employee['LastName'],
+            'full_name': normalize_full_name(employee['FirstName'], employee['LastName']),
+            'status': self.enum.EMPLOYEE_STATUS_CURRENT if employee['EmpActive'] == '1' else self.enum.EMPLOYEE_STATUS_FORMER,
+        }
+
+
 class StoreImporter(FromCOREPOSAPI, corepos_importing.model.StoreImporter):
     """
     Importer for store data from CORE POS API.
@@ -488,18 +518,29 @@ class ProductImporter(FromCOREPOSAPI, corepos_importing.model.ProductImporter):
         self.vendor_items_by_upc = {}
 
         def cache(item, i):
-            self.vendor_items_by_upc.setdefault(item['upc'], []).append(item)
+            if item.get('upc'):
+                self.vendor_items_by_upc.setdefault(item['upc'], []).append(item)
 
         self.progress_loop(cache, self.api.get_vendor_items(),
                            message="Caching CORE Vendor Items")
 
-        self.maxval_unit_size = self.app.maxval(model.Product.unit_size)
-
     def get_host_objects(self):
-        return self.api.get_products()
+        products = OrderedDict()
+
+        def collect(product, i):
+            if product.get('upc'):
+                if product['upc'] in products:
+                    log.warning("duplicate UPC encountered for '%s'; will discard: %s",
+                                product['upc'], product)
+                else:
+                    products[product['upc']] = product
+
+        self.progress_loop(collect, self.api.get_products(),
+                           message="Fetching product info from CORE-POS")
+        return list(products.values())
 
     def identify_product(self, corepos_product):
-        model = self.config.get_model()
+        model = self.app.model
         corepos_id = int(corepos_product['id'])
 
         if hasattr(self, 'core_existing'):
@@ -536,6 +577,7 @@ class ProductImporter(FromCOREPOSAPI, corepos_importing.model.ProductImporter):
         return self.api.get_vendor_items(upc=api_product['upc'])
 
     def normalize_host_object(self, product):
+        model = self.model
         if 'upc' not in product:
             log.warning("CORE-POS product has no UPC: %s", product)
             return
@@ -593,7 +635,8 @@ class ProductImporter(FromCOREPOSAPI, corepos_importing.model.ProductImporter):
                 'uom_abbreviation': (size_info['uom_abbrev'] or '').strip() or None,
             })
 
-            if data['unit_size'] and data['unit_size'] >= self.maxval_unit_size:
+            maxval = self.app.maxval(model.Product.unit_size)
+            if data['unit_size'] and data['unit_size'] >= maxval:
                 log.warning("unit_size too large (%s) for product %s, will use null instead: %s",
                             data['unit_size'], data['upc'], product)
                 data['unit_size'] = None
@@ -654,9 +697,7 @@ class ProductCostImporter(FromCOREPOSAPI, corepos_importing.model.ProductCostImp
     """
     Importer for product cost data from CORE POS API.
     """
-    # TODO: should change key after live sites are updated
-    key = ('vendor_uuid', 'code')
-    # key = ('corepos_vendor_id', 'corepos_sku')
+    key = ('corepos_vendor_id', 'corepos_sku')
     supported_fields = [
         'corepos_vendor_id',
         'corepos_sku',
@@ -671,7 +712,7 @@ class ProductCostImporter(FromCOREPOSAPI, corepos_importing.model.ProductCostImp
 
     def setup(self):
         super().setup()
-        model = self.config.get_model()
+        model = self.app.model
 
         query = self.session.query(model.Vendor)\
                             .join(model.CoreVendor)\
@@ -694,8 +735,62 @@ class ProductCostImporter(FromCOREPOSAPI, corepos_importing.model.ProductCostImp
                                                         model.Product,
                                                         key='item_id')
 
+    def should_warn_for_missing_vendor_id(self):
+        return self.config.getbool('rattail.importing.corepos.vendor_items.warn_for_missing_vendor_id',
+                                   default=True)
+
     def get_host_objects(self):
-        return self.api.get_vendor_items()
+
+        # first we will cache API products by upc
+        products = OrderedDict()
+
+        def cache(product, i):
+            if product.get('upc'):
+                products[product['upc']] = product
+
+        self.progress_loop(cache, self.api.get_products(),
+                           message="Caching product data from CORE")
+
+        # next we cache API vendor items, also by upc
+        vendor_items = {}
+        warn_for_missing_vendor_id = self.should_warn_for_missing_vendor_id()
+
+        def cache(item, i):
+            if not item.get('upc'):
+                log.warning("CORE vendor item has no upc: %s", item)
+                return
+            if item['vendorID'] == '0':
+                logger = log.warning if warn_for_missing_vendor_id else log.debug
+                logger("CORE vendor item has no vendorID: %s", item)
+                return
+            vendor_items.setdefault(item['upc'], []).append(item)
+
+        self.progress_loop(cache, self.api.get_vendor_items(),
+                           message="Caching vendor item data from CORE")
+
+        # now we must "sort" the vendor items for each upc.  to do
+        # this we just ensure the item for default vendor is first
+
+        def organize(upc, i):
+            product = products.get(upc)
+            if not product:
+                return          # product not found
+
+            vendor_id = product['default_vendor_id']
+            if not vendor_id:
+                return          # product has no default vendor
+
+            items = vendor_items[upc]
+            self.sort_these_vendor_items(items, vendor_id)
+
+        self.progress_loop(organize, list(vendor_items),
+                           message="Sorting items by default vendor")
+
+        # keep the vendor item cache for reference later
+        self.api_vendor_items = vendor_items
+
+        # host objects are the API products (in original sequence)
+        return list(products.values())
 
     def get_vendor(self, item):
         corepos_id = int(item['vendorID'])
@@ -703,7 +798,7 @@ class ProductCostImporter(FromCOREPOSAPI, corepos_importing.model.ProductCostImp
         if hasattr(self, 'vendors'):
             return self.vendors.get(corepos_id)
 
-        model = self.config.get_model()
+        model = self.app.model
         try:
             return self.session.query(model.Vendor)\
                                .join(model.CoreVendor)\
@@ -719,7 +814,9 @@ class ProductCostImporter(FromCOREPOSAPI, corepos_importing.model.ProductCostImp
         return self.api.get_product(item['upc'])
 
     def get_product(self, item):
-        item_id = item['upc']
+        item_id = item.get('upc')
+        if not item_id:
+            return
 
         if hasattr(self, 'products_by_item_id'):
             return self.products_by_item_id.get(item_id)
@@ -732,6 +829,31 @@ class ProductCostImporter(FromCOREPOSAPI, corepos_importing.model.ProductCostImp
         except orm.exc.NoResultFound:
             pass
 
+    def normalize_host_data(self, host_objects=None):
+
+        # TODO: this all seems a bit hacky but works for now..
+        # could even be we don't need this method?
+
+        if host_objects is None:
+            host_objects = self.get_host_objects()
+        normalized = []
+        self.sorted_vendor_items = {}
+
+        def normalize(product, i):
+            if not product.get('upc'):
+                log.warning("product has no upc: %s", product)
+                return
+            items = self.sort_vendor_items(product)
+            self.sorted_vendor_items[product['upc']] = items
+            for item in items:
+                data = self.normalize_host_object(item)
+                if data:
+                    normalized.append(data)
+
+        self.progress_loop(normalize, host_objects,
+                           message=f"Reading Product data from {self.host_system_title}")
+        return normalized
+
     def normalize_host_object(self, item):
         vendor = self.get_vendor(item)
         if not vendor:
@@ -751,10 +873,6 @@ class ProductCostImporter(FromCOREPOSAPI, corepos_importing.model.ProductCostImp
         #     log.warning("CORE POS product not found for item: %s", item)
         #     return
 
-        preferred = False
-        if core_product and core_product['default_vendor_id'] == item['vendorID']:
-            preferred = True
-
         case_size = decimal.Decimal(item['units'])
         unit_cost = item.get('cost')
         if unit_cost is not None:
@@ -763,7 +881,7 @@ class ProductCostImporter(FromCOREPOSAPI, corepos_importing.model.ProductCostImp
         if unit_cost is not None:
             case_cost = unit_cost * case_size
 
-        return {
+        data = {
             'corepos_vendor_id': int(item['vendorID']),
             'corepos_sku': item['sku'],
             'product_uuid': product.uuid,
@@ -772,9 +890,56 @@ class ProductCostImporter(FromCOREPOSAPI, corepos_importing.model.ProductCostImp
             'case_size': case_size,
             'case_cost': case_cost,
             'unit_cost': unit_cost,
-            'preferred': preferred,
         }
 
+        if self.fields_active(['preference', 'preferred']):
+            items = self.get_sorted_vendor_items(item)
+            i = items.index(item)
+            data['preference'] = i + 1
+            data['preferred'] = i == 0
+
+        return data
+
+    def get_sorted_vendor_items(self, item):
+        if hasattr(self, 'sorted_vendor_items'):
+            return self.sorted_vendor_items.get(item['upc'])
+
+        product = self.api.get_product(item['upc'])
+        return self.sort_vendor_items(product)
+
+    def sort_vendor_items(self, product):
+
+        # TODO: this all seems a bit hacky but works for now..
+
+        if not product.get('upc'):
+            return []
+
+        if hasattr(self, 'api_vendor_items'):
+            return self.api_vendor_items.get(product['upc'], [])
+
+        # nb. remaining logic is for real-time datasync.  here we
+        # do not have a cache of vendor items so must fetch what
+        # we need from API.  unfortunately we must (?) fetch *all*
+        # vendor items and then filter locally
+        items = [item
+                 for item in self.api.get_vendor_items()
+                 if item.get('upc') == product['upc']]
+
+        vendor_id = product['default_vendor_id']
+        self.sort_these_vendor_items(items, vendor_id)
+        return items
+
+    def sort_these_vendor_items(self, items, default_vendor_id):
+        for item in items:
+            if item['vendorID'] == default_vendor_id:
+                # found the default vendor item
+                i = items.index(item)
+                if i != 0:
+                    # it was not first; make it so
+                    items.pop(i)
+                    items.insert(0, item)
+                break
+
 
 class MembershipTypeImporter(FromCOREPOSAPI, importing.model.MembershipTypeImporter):
     """
@@ -800,14 +965,13 @@ class MemberImporter(FromCOREPOSAPI, corepos_importing.model.MemberImporter):
     """
     Importer for member data from CORE POS API.
     """
-    # TODO use this key instead
-    #key = 'corepos_card_number'
-    key = 'number'
+    key = 'corepos_card_number'
     supported_fields = [
         'number',
         'corepos_account_id',
         'corepos_card_number',
         'customer_uuid',
+        'person_uuid',
         'person_first_name',
         'person_last_name',
         'membership_type_number',
@@ -837,9 +1001,69 @@ class MemberImporter(FromCOREPOSAPI, corepos_importing.model.MemberImporter):
             model.Customer,
             key='number')
 
+        query = self.session.query(model.Person)\
+                            .outerjoin(model.Customer,
+                                       model.Customer.account_holder_uuid == model.Person.uuid)\
+                            .outerjoin(model.CoreCustomer)\
+                            .outerjoin(model.Member,
+                                       model.Member.person_uuid == model.Person.uuid)\
+                            .outerjoin(model.CoreMember)\
+                            .filter(sa.or_(
+                                model.CoreCustomer.corepos_card_number != None,
+                                model.CoreMember.corepos_card_number != None))\
+                            .options(orm.joinedload(model.Person.customer_accounts)\
+                                     .joinedload(model.Customer._corepos))
+        def card_number(person, normal):
+            customer = self.app.get_customer(person)
+            if customer and customer.corepos_card_number:
+                return customer.corepos_card_number
+            member = self.app.get_member(person)
+            if member and member.corepos_card_number:
+                return member.corepos_card_number
+        self.people_by_card_number = self.cache_model(model.Person, query=query,
+                                                      key=card_number)
+
+        self.membership_type_number_non_member = self.get_membership_type_number_non_member()
+
+    def get_membership_type_number_non_member(self):
+        if hasattr(self, 'membership_type_number_non_member'):
+            return self.membership_type_number_non_member
+
+        return self.config.getint('corepos.membership_type.non_member')
+
+    def should_warn_for_unknown_membership_type(self):
+        return self.config.getbool('rattail.importing.corepos.warn_for_unknown_membership_type',
+                                   default=True)
+
     def get_host_objects(self):
         return self.get_core_members()
 
+    def get_person(self, card_number):
+        if hasattr(self, 'people_by_card_number'):
+            return self.people_by_card_number.get(card_number)
+
+        model = self.model
+
+        try:
+            return self.session.query(model.Person)\
+                               .join(model.Customer,
+                                     model.Customer.account_holder_uuid == model.Person.uuid)\
+                               .join(model.CoreCustomer)\
+                               .filter(model.CoreCustomer.corepos_card_number == card_number)\
+                               .one()
+        except orm.exc.NoResultFound:
+            pass
+
+        try:
+            return self.session.query(model.Person)\
+                               .join(model.Member,
+                                     model.Member.person_uuid == model.Person.uuid)\
+                               .join(model.CoreMember)\
+                               .filter(model.CoreMember.corepos_card_number == card_number)\
+                               .one()
+        except orm.exc.NoResultFound:
+            pass
+
     def get_customer_by_number(self, number):
         if hasattr(self, 'customers_by_number'):
             return self.customers_by_number.get(number)
@@ -860,6 +1084,7 @@ class MemberImporter(FromCOREPOSAPI, corepos_importing.model.MemberImporter):
     def normalize_host_object(self, member):
         card_number = member['cardNo']
         customer = self.get_customer_by_number(card_number)
+        person = self.get_person(card_number)
 
         # TODO: at first i was *skipping* non-member status records,
         # but since CORE sort of assumes all customers are members,
@@ -867,8 +1092,9 @@ class MemberImporter(FromCOREPOSAPI, corepos_importing.model.MemberImporter):
         # important to import the full member info from CORE, so that
         # we have it to sync back.  therefore can't afford to "skip"
         # any member records here
-        if (member['memberStatus'] not in self.member_status_codes
-            and member['memberStatus'] not in self.non_member_status_codes):
+        memstatus = (member['memberStatus'] or '').upper() or None
+        if (memstatus not in self.member_status_codes
+            and memstatus not in self.non_member_status_codes):
             log.warning("unexpected status '%s' for member %s: %s",
                         member['memberStatus'], card_number, member)
 
@@ -893,15 +1119,24 @@ class MemberImporter(FromCOREPOSAPI, corepos_importing.model.MemberImporter):
         typeno = int(member['customerTypeID'] or 0)
         memtype = self.get_membership_type_by_number(typeno)
         if not memtype:
-            log.warning("unknown customerTypeID (membership_type_number) '%s' for: %s",
-                        member['customerTypeID'], member)
-            typeno = None
+            typeno = self.get_membership_type_number_non_member()
+            if typeno is not None:
+                memtype = self.get_membership_type_by_number(typeno)
+                if not memtype:
+                    raise ValueError("configured membership type for non-members is invalid!")
+
+            logger = log.warning if self.should_warn_for_unknown_membership_type() else log.debug
+            logger("unknown customerTypeID (membership_type_number) '%s' for: %s",
+                   member['customerTypeID'], member)
+            if typeno is not None:
+                log.debug("(will override with membership_type_number: %s)", typeno)
 
         data = {
             'number': card_number,
             'corepos_account_id': int(member['customerAccountID']),
             'corepos_card_number': card_number,
             'customer_uuid': customer.uuid if customer else None,
+            'person_uuid': person.uuid if person else None,
             'person_first_name': None,
             'person_last_name': None,
             'membership_type_number': typeno,

rattail_corepos/importing/corepos/db.py

@@ -2,7 +2,7 @@
 ################################################################################
 #
 #  Rattail -- Retail Software Framework
-#  Copyright © 2010-2023 Lance Edgar
+#  Copyright © 2010-2024 Lance Edgar
 #
 #  This file is part of Rattail.
 #
@@ -32,11 +32,7 @@ from collections import OrderedDict
 import sqlalchemy as sa
 from sqlalchemy import orm
 
-from corepos.db.office_op import model as corepos, Session as CoreSession
-from corepos.db.office_trans import model as coretrans, Session as CoreTransSession
-
 from rattail import importing
-from rattail.gpc import GPC
 from rattail.db.util import normalize_full_name
 from rattail_corepos import importing as corepos_importing
 
@@ -58,13 +54,16 @@ class FromCOREPOSToRattail(importing.FromSQLAlchemyHandler, importing.ToRattailH
         return "CORE-POS (DB/{})".format(self.corepos_dbkey)
 
     def make_host_session(self):
+        corepos = self.app.get_corepos_handler()
 
         # session type depends on the --corepos-dbtype arg
         if self.corepos_dbtype == 'office_trans':
-            return CoreTransSession(bind=self.config.coretrans_engines[self.corepos_dbkey])
+            return corepos.make_session_office_trans(
+                bind=self.config.coretrans_engines[self.corepos_dbkey])
 
         # assume office_op by default
-        return CoreSession(bind=self.config.corepos_engines[self.corepos_dbkey])
+        return corepos.make_session_office_op(
+            bind=self.config.corepos_engines[self.corepos_dbkey])
 
     def get_importers(self):
         importers = OrderedDict()
@@ -120,7 +119,13 @@ class StoreImporter(FromCOREPOS, corepos_importing.model.StoreImporter):
     """
     Importer for store data from CORE POS.
     """
-    host_model_class = corepos.Store
+
+    @property
+    def host_model_class(self):
+        corepos = self.app.get_corepos_handler()
+        op_model = corepos.get_model_office_op()
+        return op_model.Store
+
     key = 'corepos_id'
     supported_fields = [
         'corepos_id',
@@ -140,7 +145,13 @@ class EmployeeImporter(FromCOREPOS, corepos_importing.model.EmployeeImporter):
     """
     Importer for employee data from CORE POS.
     """
-    host_model_class = corepos.Employee
+
+    @property
+    def host_model_class(self):
+        corepos = self.app.get_corepos_handler()
+        op_model = corepos.get_model_office_op()
+        return op_model.Employee
+
     key = 'corepos_number'
     supported_fields = [
         'corepos_number',
@@ -164,7 +175,13 @@ class CustomerImporter(FromCOREPOS, corepos_importing.model.CustomerImporter):
     """
     Importer for customer data from CORE POS.
     """
-    host_model_class = corepos.MemberInfo
+
+    @property
+    def host_model_class(self):
+        corepos = self.app.get_corepos_handler()
+        op_model = corepos.get_model_office_op()
+        return op_model.MemberInfo
+
     key = 'corepos_card_number'
     supported_fields = [
         'corepos_card_number',
@@ -275,7 +292,13 @@ class MemberImporter(FromCOREPOS, corepos_importing.model.MemberImporter):
     """
     Importer for member data from CORE POS.
     """
-    host_model_class = corepos.MemberInfo
+
+    @property
+    def host_model_class(self):
+        corepos = self.app.get_corepos_handler()
+        op_model = corepos.get_model_office_op()
+        return op_model.MemberInfo
+
     # TODO use this key instead
     #key = 'corepos_card_number'
     key = 'number'
@@ -407,7 +430,13 @@ class TaxImporter(FromCOREPOS, corepos_importing.model.TaxImporter):
     """
     Importer for tax data from CORE POS.
     """
-    host_model_class = corepos.TaxRate
+
+    @property
+    def host_model_class(self):
+        corepos = self.app.get_corepos_handler()
+        op_model = corepos.get_model_office_op()
+        return op_model.TaxRate
+
     key = 'corepos_id'
     supported_fields = [
         'corepos_id',
@@ -429,7 +458,13 @@ class TenderImporter(FromCOREPOS, corepos_importing.model.TenderImporter):
     """
     Importer for tender data from CORE POS.
     """
-    host_model_class = corepos.Tender
+
+    @property
+    def host_model_class(self):
+        corepos = self.app.get_corepos_handler()
+        op_model = corepos.get_model_office_op()
+        return op_model.Tender
+
     key = 'corepos_id'
     supported_fields = [
         'corepos_id',
@@ -449,7 +484,13 @@ class VendorImporter(FromCOREPOS, corepos_importing.model.VendorImporter):
     """
     Importer for vendor data from CORE POS.
     """
-    host_model_class = corepos.Vendor
+
+    @property
+    def host_model_class(self):
+        corepos = self.app.get_corepos_handler()
+        op_model = corepos.get_model_office_op()
+        return op_model.Vendor
+
     key = 'corepos_id'
     supported_fields = [
         'corepos_id',
@@ -467,7 +508,7 @@ class VendorImporter(FromCOREPOS, corepos_importing.model.VendorImporter):
         """
         # can't just use rattail.db.model b/c the CoreVendor would normally not
         # be in there!  this still requires custom model to be configured though.
-        model = self.config.get_model()
+        model = self.app.model
 
         # first get default query
         query = super().cache_query()
@@ -502,7 +543,13 @@ class DepartmentImporter(FromCOREPOS, corepos_importing.model.DepartmentImporter
     """
     Importer for department data from CORE POS.
     """
-    host_model_class = corepos.Department
+
+    @property
+    def host_model_class(self):
+        corepos = self.app.get_corepos_handler()
+        op_model = corepos.get_model_office_op()
+        return op_model.Department
+
     key = 'corepos_number'
     supported_fields = [
         'corepos_number',
@@ -526,7 +573,13 @@ class SubdepartmentImporter(FromCOREPOS, corepos_importing.model.SubdepartmentIm
     """
     Importer for subdepartment data from CORE POS.
     """
-    host_model_class = corepos.Subdepartment
+
+    @property
+    def host_model_class(self):
+        corepos = self.app.get_corepos_handler()
+        op_model = corepos.get_model_office_op()
+        return op_model.Subdepartment
+
     key = 'corepos_number'
     supported_fields = [
         'corepos_number',
@@ -548,7 +601,13 @@ class ProductImporter(FromCOREPOS, corepos_importing.model.ProductImporter):
     """
     Importer for product data from CORE POS.
     """
-    host_model_class = corepos.Product
+
+    @property
+    def host_model_class(self):
+        corepos = self.app.get_corepos_handler()
+        op_model = corepos.get_model_office_op()
+        return op_model.Product
+
     key = 'corepos_id'
     supported_fields = [
         'corepos_id',
@@ -579,6 +638,8 @@ class ProductImporter(FromCOREPOS, corepos_importing.model.ProductImporter):
 
     def setup(self):
         super().setup()
+        corepos = self.app.get_corepos_handler()
+        op_model = corepos.get_model_office_op()
 
         if self.fields_active(self.sale_price_fields):
             self.core_batch_items = {}
@@ -588,11 +649,11 @@ class ProductImporter(FromCOREPOS, corepos_importing.model.ProductImporter):
             # determine which would "win" but not clear what sort
             # order should be used, e.g. CORE does not seem to use one
             today = self.app.today()
-            batches = self.host_session.query(corepos.Batch)\
-                                       .filter(corepos.Batch.start_date <= today)\
-                                       .filter(corepos.Batch.end_date >= today)\
-                                       .filter(corepos.Batch.discount_type > 0)\
-                                       .options(orm.joinedload(corepos.Batch.items))\
+            batches = self.host_session.query(op_model.Batch)\
+                                       .filter(op_model.Batch.start_date <= today)\
+                                       .filter(op_model.Batch.end_date >= today)\
+                                       .filter(op_model.Batch.discount_type > 0)\
+                                       .options(orm.joinedload(op_model.Batch.items))\
                                        .all()
 
             def cache(batch, i):
@@ -620,7 +681,7 @@ class ProductImporter(FromCOREPOS, corepos_importing.model.ProductImporter):
     def normalize_host_object(self, product):
 
         try:
-            upc = GPC(product.upc, calc_check_digit='upc')
+            upc = self.app.make_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':
@@ -691,7 +752,13 @@ class ProductCostImporter(FromCOREPOS, corepos_importing.model.ProductCostImport
     """
     Importer for product cost data from CORE POS API.
     """
-    host_model_class = corepos.VendorItem
+
+    @property
+    def host_model_class(self):
+        corepos = self.app.get_corepos_handler()
+        op_model = corepos.get_model_office_op()
+        return op_model.VendorItem
+
     key = ('corepos_vendor_id', 'corepos_sku')
     supported_fields = [
         'corepos_vendor_id',
@@ -719,9 +786,12 @@ class ProductCostImporter(FromCOREPOS, corepos_importing.model.ProductCostImport
         self.products_by_item_id = self.cache_model(model.Product, key='item_id')
 
     def query(self):
+        corepos = self.app.get_corepos_handler()
+        op_model = corepos.get_model_office_op()
+
         query = super().query()
 
-        query = query.options(orm.joinedload(corepos.VendorItem.product))
+        query = query.options(orm.joinedload(op_model.VendorItem.product))
 
         return query
 
@@ -731,7 +801,7 @@ class ProductCostImporter(FromCOREPOS, corepos_importing.model.ProductCostImport
         if hasattr(self, 'vendors_by_corepos_id'):
             return self.vendors_by_corepos_id.get(corepos_id)
 
-        model = self.config.get_model()
+        model = self.app.model
         try:
             return self.session.query(model.Vendor)\
                                .join(model.CoreVendor)\
@@ -798,7 +868,13 @@ class MemberEquityPaymentImporter(FromCOREPOS, corepos_importing.model.MemberEqu
     """
     Imports equity payment data from CORE-POS
     """
-    host_model_class = coretrans.StockPurchase
+
+    @property
+    def host_model_class(self):
+        corepos = self.app.get_corepos_handler()
+        trans_model = corepos.get_model_office_trans()
+        return trans_model.StockPurchase
+
     key = 'uuid'
     supported_fields = [
         'uuid',
@@ -877,7 +953,17 @@ class MemberEquityPaymentImporter(FromCOREPOS, corepos_importing.model.MemberEqu
         if len(match) == 1:
             return match[0]
 
-        # nb. avoid datetime for this one
+        # then try to match on date only, not time
+        match = [payment for payment in payments
+                 if payment.corepos_transaction_number == stock_purchase.transaction_number
+                 and payment.corepos_transaction_id == stock_purchase.transaction_id
+                 and payment.amount == stock_purchase.amount
+                 and payment.corepos_department_number == stock_purchase.department_number
+                 and self.app.localtime(payment.corepos_datetime, from_utc=True).date() == dt.date()]
+        if len(match) == 1:
+            return match[0]
+
+        # nb. avoid date/time for this one
         matches = [payment for payment in payments
                    if payment.corepos_transaction_number == stock_purchase.transaction_number
                    and payment.corepos_transaction_id == stock_purchase.transaction_id
@@ -890,6 +976,9 @@ class MemberEquityPaymentImporter(FromCOREPOS, corepos_importing.model.MemberEqu
                         stock_purchase.amount,
                         stock_purchase.datetime)
 
+            # TODO: now that we try to match on date above, this logic
+            # may no longer be necssary/useful?
+
             # so there is one match, but its timestamp may be way off,
             # so let's also make sure at least date matches
             payment = matches[0]

rattail_corepos/importing/model.py

@@ -2,7 +2,7 @@
 ################################################################################
 #
 #  Rattail -- Retail Software Framework
-#  Copyright © 2010-2023 Lance Edgar
+#  Copyright © 2010-2024 Lance Edgar
 #
 #  This file is part of Rattail.
 #
@@ -27,7 +27,6 @@ Rattail model importer extensions, for CORE-POS integration
 import decimal
 
 from rattail import importing
-from rattail.util import pretty_quantity
 
 
 ##############################
@@ -44,7 +43,7 @@ class PersonImporter(importing.model.PersonImporter):
 
     def cache_query(self):
         query = super().cache_query()
-        model = self.config.get_model()
+        model = self.app.model
 
         # we want to ignore people with no CORE ID, if that's (part of) our key
         if 'corepos_customer_id' in self.key:
@@ -177,7 +176,7 @@ class ProductImporter(importing.model.ProductImporter):
 
     def cache_query(self):
         query = super().cache_query()
-        model = self.config.get_model()
+        model = self.app.model
 
         # we want to ignore products with no CORE ID, if that's (part of) our key
         if 'corepos_id' in self.key:
@@ -219,9 +218,10 @@ class ProductImporter(importing.model.ProductImporter):
             uom_code = self.get_uom_code(uom_abbrev) or self.enum.UNIT_OF_MEASURE_NONE
 
         if unit_size is not None and uom_abbrev is not None:
-            size = "{} {}".format(pretty_quantity(unit_size), uom_abbrev)
+            size = self.app.render_quantity(unit_size)
+            size = f"{size} {uom_abbrev}"
         elif unit_size is not None:
-            size = pretty_quantity(unit_size)
+            size = self.app.render_quantity(unit_size)
         elif uom_abbrev is not None:
             size = uom_abbrev
         else:
@@ -247,7 +247,7 @@ class ProductCostImporter(importing.model.ProductCostImporter):
 
     def cache_query(self):
         query = super().cache_query()
-        model = self.config.get_model()
+        model = self.app.model
 
         # we want to ignore items with no CORE ID, if that's (part of) our key
         if 'corepos_id' in self.key:

rattail_corepos/importing/versions.py

@@ -2,7 +2,7 @@
 ################################################################################
 #
 #  Rattail -- Retail Software Framework
-#  Copyright © 2010-2023 Lance Edgar
+#  Copyright © 2010-2024 Lance Edgar
 #
 #  This file is part of Rattail.
 #
@@ -51,7 +51,7 @@ class CorePersonImporter(base.VersionImporter):
 
     @property
     def host_model_class(self):
-        model = self.config.get_model()
+        model = self.app.model
         return model.CorePerson
 
 
@@ -59,7 +59,7 @@ class CoreEmployeeImporter(base.VersionImporter):
 
     @property
     def host_model_class(self):
-        model = self.config.get_model()
+        model = self.app.model
         return model.CoreEmployee
 
 
@@ -67,7 +67,7 @@ class CoreCustomerImporter(base.VersionImporter):
 
     @property
     def host_model_class(self):
-        model = self.config.get_model()
+        model = self.app.model
         return model.CoreCustomer
 
 
@@ -82,7 +82,7 @@ class CoreMemberImporter(base.VersionImporter):
 
     @property
     def host_model_class(self):
-        model = self.config.get_model()
+        model = self.app.model
         return model.CoreMember
 
 
@@ -90,7 +90,7 @@ class CoreMemberEquityPaymentImporter(base.VersionImporter):
 
     @property
     def host_model_class(self):
-        model = self.config.get_model()
+        model = self.app.model
         return model.CoreMemberEquityPayment
 
 
@@ -98,7 +98,7 @@ class CoreStoreImporter(base.VersionImporter):
 
     @property
     def host_model_class(self):
-        model = self.config.get_model()
+        model = self.app.model
         return model.CoreStore
 
 
@@ -106,7 +106,7 @@ class CoreDepartmentImporter(base.VersionImporter):
 
     @property
     def host_model_class(self):
-        model = self.config.get_model()
+        model = self.app.model
         return model.CoreDepartment
 
 
@@ -114,7 +114,7 @@ class CoreSubdepartmentImporter(base.VersionImporter):
 
     @property
     def host_model_class(self):
-        model = self.config.get_model()
+        model = self.app.model
         return model.CoreSubdepartment
 
 
@@ -122,7 +122,7 @@ class CoreVendorImporter(base.VersionImporter):
 
     @property
     def host_model_class(self):
-        model = self.config.get_model()
+        model = self.app.model
         return model.CoreVendor
 
 
@@ -130,5 +130,5 @@ class CoreProductImporter(base.VersionImporter):
 
     @property
     def host_model_class(self):
-        model = self.config.get_model()
+        model = self.app.model
         return model.CoreProduct