Remove deferred FK validation API and unused backend bulk helpers

The streaming deserialization pipeline (#392) replaced the only callers of
these. Removes SyncableModel.deferred_clean_fields, ForeignKeyReference,
TemporaryTable, and the backend _bulk_full_record_upsert and _bulk_update
helpers, along with the helpers that only they used: _bulk_insert,
_create_temporary_table, _create_placeholder_list, the PostgreSQL
_prepare_* methods and get_pk_field.

calculate_max_sqlite_variables is kept because Kolibri imports it.

Closes #396

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Author: bjester

CHANGELOG.md

@@ -11,6 +11,9 @@ List of the most important changes for each release.
 - Children of a dirty parent that is left out of the deserialization run are now deserialized against the parent's existing app row
 - For models whose `clean_fields` loads related objects, a dirty self-referential chain is now deserialized one level per run
 - Logs deserialization warnings under `morango.sync.stream.deserialize` instead of `morango.sync.operations`
+- Removes `SyncableModel.deferred_clean_fields` and `morango.models.core.ForeignKeyReference`
+- Removes `morango.sync.backends.utils.TemporaryTable` and `morango.sync.backends.utils.get_pk_field`
+- Removes the unused bulk write helpers `_bulk_full_record_upsert`, `_bulk_update` and `_bulk_insert`, and the temporary table helper `_create_temporary_table`, from the sync backends
 
 ## 0.8.15
 - Improves buffer serialization performance by performing bulk counter lookup

morango/models/core.py

@@ -2,8 +2,6 @@
 import json
 import logging
 import uuid
-from collections import defaultdict
-from collections import namedtuple
 from functools import reduce
 
 from django.core import exceptions
@@ -745,9 +743,6 @@ class RecordMaxCounterBuffer(AbstractCounter):
     model_uuid = UUIDField(db_index=True)
 
 
-ForeignKeyReference = namedtuple("ForeignKeyReference", ["from_field", "from_pk", "to_pk"])
-
-
 class SyncableModel(UUIDModelMixin):
     """
     ``SyncableModel`` is the base model class for syncing. Other models inherit from this class if they want to make
@@ -873,39 +868,6 @@ def cached_clean_fields(self, fk_lookup_cache, exclude=None, sync_filter=None):
         )
         fk_lookup_cache[self_key] = 1
 
-    def deferred_clean_fields(self, exclude=None, sync_filter=None):
-        """
-        Calls `.clean_fields()` but excludes all foreign key fields and instead returns them as a
-        dictionary for deferred batch processing
-
-        :param exclude: A list of field names to exclude from validation
-        :type exclude: list[str]
-        :param sync_filter: The current sync's filter, if any
-        :type sync_filter: Filter|None
-        :return: A dictionary containing lists of `ForeignKeyReference`s keyed by the name of the
-            model being referenced by the FK
-        """
-        excluded_fields = exclude or []
-        deferred_fks = defaultdict(list)
-        for field in self._meta.fields:
-            if not isinstance(field, models.ForeignKey):
-                continue
-            # by not excluding the field if it's null, the default validation logic will apply
-            # and should raise a ValidationError if the FK field is not nullable
-            if getattr(self, field.attname) is None:
-                continue
-            excluded_fields.append(field.name)
-            deferred_fks[field.related_model._meta.verbose_name].append(
-                ForeignKeyReference(
-                    from_field=field.attname,
-                    from_pk=self.pk,
-                    to_pk=getattr(self, field.attname),
-                )
-            )
-
-        self.clean_fields(exclude=excluded_fields, sync_filter=sync_filter)
-        return deferred_fks
-
     def serialize(self):
         """All concrete fields of the ``SyncableModel`` subclass, except for those specifically blacklisted, are returned in a dict."""
         # NOTE: code adapted from https://github.com/django/django/blob/master/django/forms/models.py#L75

morango/sync/backends/base.py

@@ -7,8 +7,6 @@
 
 
 class BaseSQLWrapper(object):
-    create_temporary_table_template = "CREATE TEMP TABLE {name} ({fields})"
-
     def __init__(self, connection):
         self.connection = connection
 
@@ -25,31 +23,6 @@ def _set_transaction_repeatable_read(self):
         """Set the current transaction isolation level"""
         yield
 
-    def _create_placeholder_list(self, fields, db_values):
-        # number of rows to update
-        num_of_rows = len(db_values) // len(fields)
-        # create '%s' placeholders for a single row
-        placeholder_tuple = tuple(["%s" for _ in range(len(fields))])
-        # create list of the '%s' tuple placeholders based on number of rows to update
-        return [str(placeholder_tuple) for _ in range(num_of_rows)]
-
-    def _bulk_full_record_upsert(self, cursor, table_name, fields, db_values):
-        raise NotImplementedError("Subclass must implement this method.")
-
-    def _bulk_insert(self, cursor, table_name, fields, db_values):
-        placeholder_str = ", ".join(self._create_placeholder_list(fields, db_values)).replace(
-            "'", ""
-        )
-        fields_str = str(tuple(str(f.attname) for f in fields)).replace("'", "")
-        insert = """
-            INSERT INTO {table_name} {fields}
-            VALUES {placeholder_str}
-        """.format(table_name=table_name, fields=fields_str, placeholder_str=placeholder_str)
-        cursor.execute(insert, db_values)
-
-    def _bulk_update(self, cursor, table_name, fields, db_values):
-        raise NotImplementedError("Subclass must implement this method.")
-
     def _dequeuing_delete_rmcb_records(self, cursor, transfersession_id):
         # delete all RMCBs which are a reverse FF (store version newer than buffer version)
         delete_rmcb_records = """DELETE FROM {rmcb}
@@ -176,16 +149,6 @@ def _dequeuing_delete_remaining_buffer(self, cursor, transfersession_id):
         )
         cursor.execute(delete_remaining_buffer)
 
-    def _create_temporary_table(self, cursor, name, field_sqls, fields_params):
-        """
-        :param cursor: The database connection cursor
-        :param name: The str name of the temp table
-        :param field_sqls: A list of SQL strings representing the fields
-        :param fields_params: A list of SQL parameters if necessary for the fields SQL
-        """
-        sql = self.create_temporary_table_template.format(name=name, fields=", ".join(field_sqls))
-        cursor.execute(sql, fields_params)
-
     def _lock_all_partitions(self, shared=False):
         """
         Execute a lock within the database for all partitions, if the backend supports it. The lock

morango/sync/backends/postgres.py

@@ -10,7 +10,6 @@
 from morango.utils import SETTINGS
 
 from .base import BaseSQLWrapper
-from .utils import get_pk_field
 
 # advisory lock integers for locking partitions
 LOCK_ALL_PARTITIONS = 1
@@ -23,7 +22,6 @@
 
 class SQLWrapper(BaseSQLWrapper):
     backend = "postgresql"
-    create_temporary_table_template = "CREATE TEMP TABLE {name} ({fields}) ON COMMIT DROP"
 
     def _is_transaction_isolation_error(self, error):
         """
@@ -85,96 +83,6 @@ def _set_transaction_repeatable_read(self):
                     self.connection.connection.set_session(isolation_level=existing_isolation_level)
                 self.connection.connection.set_session(autocommit=existing_autocommit)
 
-    def _prepare_with_values(self, name, fields, db_values):
-        placeholder_list = self._create_placeholder_list(fields, db_values)
-        # convert this list to a string to be passed into raw sql query
-        placeholder_str = ", ".join(placeholder_list).replace("'", "")
-        return """
-            WITH {name} {fields} as
-            (
-                VALUES {placeholder_str}
-            )
-        """.format(
-            name=name,
-            fields=str(tuple(str(f.column) for f in fields)).replace("'", ""),
-            placeholder_str=placeholder_str,
-        )
-
-    def _prepare_casted_fields(self, fields):
-        return ", ".join(
-            map(
-                lambda f: "{f}::{type}".format(f=f.column, type=f.rel_db_type(self.connection)),
-                fields,
-            )
-        )
-
-    def _prepare_set_casted_values(self, fields, source_table):
-        return ", ".join(
-            map(
-                lambda f: "{f} = {src}.{f}::{type}".format(
-                    f=f.attname,
-                    type=f.rel_db_type(self.connection),
-                    src=source_table,
-                ),
-                fields,
-            )
-        )
-
-    def _bulk_full_record_upsert(self, cursor, table_name, fields, db_values):
-        pk = get_pk_field(fields)
-
-        cte_name = "new_values"
-        upsert = """
-            {cte},
-            updated as
-            (
-                UPDATE {table_name} model
-                SET {set_values}
-                FROM {cte_name} cte
-                WHERE model.id = cte.{pk_field}::{pk_type}
-                RETURNING model.{pk_field}
-            )
-            INSERT INTO {table_name} {fields}
-            SELECT {select_fields}
-            FROM {cte_name} cte
-            WHERE cte.{pk_field}::{pk_type} NOT IN (SELECT {pk_field} FROM updated)
-        """.format(
-            cte=self._prepare_with_values(cte_name, fields, db_values),
-            cte_name=cte_name,
-            table_name=table_name,
-            fields=str(tuple(str(f.column) for f in fields)).replace("'", ""),
-            set_values=self._prepare_set_casted_values(fields, "cte"),
-            select_fields=self._prepare_casted_fields(fields),
-            pk_field=pk.column,
-            pk_type=pk.rel_db_type(self.connection),
-        )
-        # use DB-APIs parameter substitution (2nd parameter expects a sequence)
-        cursor.execute(upsert, db_values)
-
-    def _bulk_update(self, cursor, table_name, fields, db_values):
-        pk = get_pk_field(fields)
-
-        insert = """
-            {cte}
-            UPDATE {table_name} model
-            SET {set_values}
-            FROM {cte_name} cte
-            WHERE model.{pk_field} = cte.{pk_field}::{pk_type}
-        """
-
-        cte_name = "new_values"
-        insert = insert.format(
-            cte=self._prepare_with_values(cte_name, fields, db_values),
-            cte_name=cte_name,
-            table_name=table_name,
-            fields=str(tuple(str(f.column) for f in fields)).replace("'", ""),
-            set_values=self._prepare_set_casted_values(fields, "cte"),
-            pk_field=pk.column,
-            pk_type=pk.rel_db_type(self.connection),
-        )
-        # use DB-APIs parameter substitution (2nd parameter expects a sequence)
-        cursor.execute(insert, db_values)
-
     def _dequeuing_merge_conflict_rmcb(self, cursor, transfersession_id):
         # transfer record max counters for records with merge conflicts + perform max
         merge_conflict_rmc = """UPDATE {rmc} as rmc SET counter

morango/sync/backends/sqlite.py

@@ -4,104 +4,11 @@
 from morango.models.core import Store
 
 from .base import BaseSQLWrapper
-from .utils import calculate_max_sqlite_variables
-from .utils import get_pk_field
 
 
 class SQLWrapper(BaseSQLWrapper):
     backend = "sqlite"
 
-    def _bulk_full_record_upsert(self, cursor, table_name, fields, db_values):
-        """
-        Example query:
-        `REPLACE INTO model (F1,F2,F3) VALUES (%s, %s, %s), (%s, %s, %s), (%s, %s, %s)`
-        where values=[1,2,3,4,5,6,7,8,9]
-        """
-        placeholder_list = self._create_placeholder_list(fields, db_values)
-        # calculate and create equal sized chunks of data to insert incrementally
-        num_of_rows_able_to_insert = calculate_max_sqlite_variables() // len(fields)
-        num_of_values_able_to_insert = num_of_rows_able_to_insert * len(fields)
-        value_chunks = [
-            db_values[x : x + num_of_values_able_to_insert]
-            for x in range(0, len(db_values), num_of_values_able_to_insert)
-        ]
-        placeholder_chunks = [
-            placeholder_list[x : x + num_of_rows_able_to_insert]
-            for x in range(0, len(placeholder_list), num_of_rows_able_to_insert)
-        ]
-        # insert data chunks
-        fields_str = str(tuple(str(f.attname) for f in fields)).replace("'", "")
-        for values, params in zip(value_chunks, placeholder_chunks):
-            placeholder_str = ", ".join(params).replace("'", "")
-            insert = """
-                REPLACE INTO {table_name} {fields}
-                VALUES {placeholder_str}
-            """.format(
-                table_name=table_name,
-                fields=fields_str,
-                placeholder_str=placeholder_str,
-            )
-            # use DB-APIs parameter substitution (2nd parameter expects a sequence)
-            cursor.execute(insert, values)
-
-    def _bulk_insert(self, cursor, table_name, fields, db_values):
-        num_of_rows_able_to_insert = calculate_max_sqlite_variables() // len(fields)
-        num_of_values_able_to_insert = num_of_rows_able_to_insert * len(fields)
-        value_chunks = [
-            db_values[x : x + num_of_values_able_to_insert]
-            for x in range(0, len(db_values), num_of_values_able_to_insert)
-        ]
-        for value_chunk in value_chunks:
-            super(SQLWrapper, self)._bulk_insert(cursor, table_name, fields, value_chunk)
-
-    def _bulk_update(self, cursor, table_name, fields, db_values):
-        """
-        Example query:
-        `UPDATE model SET F1=(CASE id WHEN %s THEN %s ... END) ...`
-        WHERE id IN [...]
-        """
-        # calculate and create equal sized chunks of data to update incrementally
-        # for every field we're updating, we'll require 3 parameters
-        num_update_fields = len(fields) - 1
-        num_of_rows_able_to_update = calculate_max_sqlite_variables() // num_update_fields // 3
-        num_of_values_able_to_update = num_of_rows_able_to_update * len(fields)
-        value_chunks = [
-            db_values[x : x + num_of_values_able_to_update]
-            for x in range(0, len(db_values), num_of_values_able_to_update)
-        ]
-        pk = get_pk_field(fields)
-
-        # insert data chunks
-        for values in value_chunks:
-            set_sql = ""
-            params = []
-            pk_params = []
-            for field in fields:
-                if field == pk:
-                    continue
-                set_field_sql = " {field} = (CASE {pk_field}".format(
-                    field=field.column, pk_field=pk.column
-                )
-                for y in range(0, len(values), len(fields)):
-                    value_set = values[y : y + len(fields)]
-                    set_field_sql += " WHEN %s THEN %s"
-                    pk_params.append(value_set[fields.index(pk)])
-                    params.append(value_set[fields.index(pk)])
-                    params.append(value_set[fields.index(field)])
-                set_field_sql += " END),"
-                set_sql += set_field_sql
-            params.extend(pk_params)
-            update = """
-                UPDATE {table_name} SET {set_sql} WHERE {pk_field} IN {placeholder_str}
-            """.format(
-                table_name=table_name,
-                set_sql=set_sql[:-1],
-                pk_field=pk.column,
-                placeholder_str="({})".format(",".join("%s" for _ in range(len(pk_params)))),
-            )
-            # use DB-APIs parameter substitution (2nd parameter expects a sequence)
-            cursor.execute(update, params)
-
     def _dequeuing_merge_conflict_rmcb(self, cursor, transfersession_id):
         # transfer record max counters for records with merge conflicts + perform max
         merge_conflict_rmc = """REPLACE INTO {rmc} (instance_id, counter, store_model_id)