perf: bulk membership creation with INSERT ON CONFLICT DO NOTHING

Replace per-record ORM creates and Command.create() tuples with raw SQL
INSERT ... ON CONFLICT (unique_cols) DO NOTHING for bulk membership
creation. Duplicates are silently skipped and the inserted count is
returned via cursor.rowcount.

Updates _import_registrants and _add_beneficiaries to use the new
skip_duplicates path, with ORM cache invalidation after raw SQL inserts.

Author: kneckinator

spp_programs/__manifest__.py

@@ -4,7 +4,7 @@
     "name": "OpenSPP Programs",
     "summary": "Manage programs, cycles, beneficiary enrollment, entitlements (cash and in-kind), payments, and fund tracking for social protection.",
     "category": "OpenSPP/Core",
-    "version": "19.0.2.0.6",
+    "version": "19.0.2.0.7",
     "sequence": 1,
     "author": "OpenSPP.org",
     "website": "https://github.com/OpenSPP/OpenSPP2",

spp_programs/models/cycle_membership.py

@@ -1,7 +1,11 @@
 # Part of OpenSPP. See LICENSE file for full copyright and licensing details.
-from odoo import _, fields, models
+import logging
+
+from odoo import _, api, fields, models
 from odoo.exceptions import ValidationError
 
+_logger = logging.getLogger(__name__)
+
 
 class SPPCycleMembership(models.Model):
     _name = "spp.cycle.membership"
@@ -87,6 +91,72 @@ def open_registrant_form(self):
                 },
             }
 
+    @api.model
+    def bulk_create_memberships(self, vals_list, chunk_size=1000, skip_duplicates=False):
+        """Create cycle memberships in bulk with optional duplicate skipping.
+
+        :param vals_list: List of dicts with membership values
+        :param chunk_size: Number of records per batch (default 1000)
+        :param skip_duplicates: When True, use INSERT ... ON CONFLICT DO NOTHING
+            to silently skip duplicate (partner_id, cycle_id) pairs.
+            Returns the count of inserted rows.
+        :return: Recordset (skip_duplicates=False) or int count (skip_duplicates=True)
+        """
+        if not vals_list:
+            return 0 if skip_duplicates else self.env["spp.cycle.membership"]
+
+        if skip_duplicates:
+            return self._bulk_insert_on_conflict(vals_list, chunk_size)
+
+        return self.create(vals_list)
+
+    def _bulk_insert_on_conflict(self, vals_list, chunk_size=1000):
+        """Insert cycle memberships using raw SQL with ON CONFLICT DO NOTHING.
+
+        :param vals_list: List of dicts with at least partner_id, cycle_id, state
+        :param chunk_size: Number of records per SQL INSERT batch
+        :return: Total number of rows actually inserted
+        """
+        cr = self.env.cr
+        uid = self.env.uid
+        total_inserted = 0
+
+        for i in range(0, len(vals_list), chunk_size):
+            batch = vals_list[i : i + chunk_size]
+            values = []
+            params = []
+            for v in batch:
+                values.append("(%s, %s, %s, %s, %s, %s, now(), now())")
+                params.extend(
+                    [
+                        v["partner_id"],
+                        v["cycle_id"],
+                        v.get("state", "draft"),
+                        v.get("enrollment_date", fields.Date.today()),
+                        uid,
+                        uid,
+                    ]
+                )
+
+            sql = """
+                INSERT INTO spp_cycle_membership
+                    (partner_id, cycle_id, state, enrollment_date,
+                     create_uid, write_uid, create_date, write_date)
+                VALUES {}
+                ON CONFLICT (partner_id, cycle_id) DO NOTHING
+            """.format(  # noqa: S608  # nosec B608
+                ", ".join(values)
+            )
+            cr.execute(sql, params)
+            total_inserted += cr.rowcount
+
+        _logger.info(
+            "Bulk inserted %d cycle memberships (%d skipped as duplicates)",
+            total_inserted,
+            len(vals_list) - total_inserted,
+        )
+        return total_inserted
+
     def unlink(self):
         if not self:
             return

spp_programs/models/managers/cycle_manager_base.py

@@ -835,25 +835,26 @@ def _add_beneficiaries(self, cycle, beneficiaries, state="draft", do_count=False
         """Add Beneficiaries
 
         :param cycle: Recordset of cycle
-        :param beneficiaries: Recordset of beneficiaries
+        :param beneficiaries: List of partner IDs
         :param state: String state to be set to beneficiary
         :param do_count: Boolean - set to False to not run compute functions
-        :return: Integer - count of not enrolled members
-        """
-        new_beneficiaries = []
-        for r in beneficiaries:
-            new_beneficiaries.append(
-                [
-                    0,
-                    0,
-                    {
-                        "partner_id": r,
-                        "enrollment_date": fields.Date.today(),
-                        "state": state,
-                    },
-                ]
-            )
-        cycle.update({"cycle_membership_ids": new_beneficiaries})
+        :return: Integer - count of inserted members
+        """
+        today = fields.Date.today()
+        vals_list = [
+            {
+                "partner_id": partner_id,
+                "cycle_id": cycle.id,
+                "enrollment_date": today,
+                "state": state,
+            }
+            for partner_id in beneficiaries
+        ]
+        self.env["spp.cycle.membership"].bulk_create_memberships(vals_list, skip_duplicates=True)
+
+        # Raw SQL bypasses the ORM cache — invalidate so subsequent reads
+        # (e.g. cycle.cycle_membership_ids) reflect the new rows.
+        cycle.invalidate_recordset(["cycle_membership_ids"])
 
         if do_count:
             # Update Statistics

spp_programs/models/managers/eligibility_manager.py

@@ -1,7 +1,7 @@
 # Part of OpenSPP. See LICENSE file for full copyright and licensing details.
 import logging
 
-from odoo import Command, _, api, fields, models
+from odoo import _, api, fields, models
 
 from odoo.addons.job_worker.delay import group
 
@@ -174,11 +174,13 @@ def mark_import_as_done(self):
 
     def _import_registrants(self, new_beneficiaries, state="draft", do_count=False):
         _logger.info("Importing %s beneficiaries", len(new_beneficiaries))
-        _logger.info("updated")
-        beneficiaries_val = []
-        for beneficiary in new_beneficiaries:
-            beneficiaries_val.append(Command.create({"partner_id": beneficiary.id, "state": state}))
-        self.program_id.update({"program_membership_ids": beneficiaries_val})
+        vals_list = [{"partner_id": b.id, "program_id": self.program_id.id, "state": state} for b in new_beneficiaries]
+        count = self.env["spp.program.membership"].bulk_create_memberships(vals_list, skip_duplicates=True)
+        _logger.info("Imported %d new memberships (%d duplicates skipped)", count, len(vals_list) - count)
+
+        # Raw SQL bypasses the ORM cache — invalidate so subsequent reads
+        # (e.g. program.program_membership_ids) reflect the new rows.
+        self.program_id.invalidate_recordset(["program_membership_ids"])
 
         if do_count:
             # Compute Statistics

spp_programs/models/program_membership.py

@@ -1,4 +1,5 @@
 # Part of OpenSPP. See LICENSE file for full copyright and licensing details.
+import logging
 
 from lxml import etree
 
@@ -7,6 +8,8 @@
 
 from . import constants
 
+_logger = logging.getLogger(__name__)
+
 
 class SPPProgramMembership(models.Model):
     _inherit = [
@@ -345,26 +348,26 @@ def action_exit(self):
             }
         )
 
-    @api.model_create_multi
-    def bulk_create_memberships(self, vals_list, chunk_size=1000):
+    @api.model
+    def bulk_create_memberships(self, vals_list, chunk_size=1000, skip_duplicates=False):
         """Create program memberships in bulk with optional chunking.
 
         This helper is intended for large enrollment jobs (e.g. CEL-driven
         bulk enrollment) where thousands of memberships need to be created
         in a single operation.
 
-        It preserves the normal create() semantics, including:
-        - standard ORM validations and constraints
-        - audit logging (via spp_audit rules)
-        - source tracking mixins
-
-        The only optimisation is to:
-        - accept already-prepared value dicts
-        - optionally split very large batches into smaller chunks to keep
-          memory use and per-transaction work bounded.
+        :param vals_list: List of dicts with membership values
+        :param chunk_size: Number of records per batch (default 1000)
+        :param skip_duplicates: When True, use INSERT ... ON CONFLICT DO NOTHING
+            to silently skip duplicate (partner_id, program_id) pairs instead of
+            raising IntegrityError. Returns the count of inserted rows.
+        :return: Recordset (skip_duplicates=False) or int count (skip_duplicates=True)
         """
         if not vals_list:
-            return self.env["spp.program.membership"]
+            return 0 if skip_duplicates else self.env["spp.program.membership"]
+
+        if skip_duplicates:
+            return self._bulk_insert_on_conflict(vals_list, chunk_size)
 
         if chunk_size and chunk_size > 0:
             all_memberships = self.env["spp.program.membership"]
@@ -386,3 +389,53 @@ def bulk_create_memberships(self, vals_list, chunk_size=1000):
             SPPProgramMembership,
             self.sudo(),  # nosemgrep: odoo-sudo-without-context
         ).create(vals_list)
+
+    def _bulk_insert_on_conflict(self, vals_list, chunk_size=1000):
+        """Insert memberships using raw SQL with ON CONFLICT DO NOTHING.
+
+        Bypasses ORM for maximum throughput during bulk enrollment. Duplicates
+        (matching the UNIQUE constraint on partner_id, program_id) are silently
+        skipped.
+
+        :param vals_list: List of dicts with at least partner_id, program_id, state
+        :param chunk_size: Number of records per SQL INSERT batch
+        :return: Total number of rows actually inserted
+        """
+        cr = self.env.cr
+        uid = self.env.uid
+        total_inserted = 0
+
+        for i in range(0, len(vals_list), chunk_size):
+            batch = vals_list[i : i + chunk_size]
+            values = []
+            params = []
+            for v in batch:
+                values.append("(%s, %s, %s, %s, %s, now(), now())")
+                params.extend(
+                    [
+                        v["partner_id"],
+                        v["program_id"],
+                        v.get("state", "draft"),
+                        uid,
+                        uid,
+                    ]
+                )
+
+            sql = """
+                INSERT INTO spp_program_membership
+                    (partner_id, program_id, state,
+                     create_uid, write_uid, create_date, write_date)
+                VALUES {}
+                ON CONFLICT (partner_id, program_id) DO NOTHING
+            """.format(  # noqa: S608  # nosec B608
+                ", ".join(values)
+            )
+            cr.execute(sql, params)
+            total_inserted += cr.rowcount
+
+        _logger.info(
+            "Bulk inserted %d program memberships (%d skipped as duplicates)",
+            total_inserted,
+            len(vals_list) - total_inserted,
+        )
+        return total_inserted

spp_programs/tests/__init__.py

@@ -32,3 +32,4 @@
 from . import test_payment_and_accounting
 from . import test_managers
 from . import test_cycle_auto_approve_fund_check
+from . import test_bulk_membership