Commenting and fixes to FK validation

Author: bjester

morango/models/core.py

@@ -4,11 +4,10 @@
 import json
 import logging
 import uuid
-from functools import reduce
 
-from functools import reduce
-from collections import namedtuple
 from collections import defaultdict
+from collections import namedtuple
+from functools import reduce
 
 from django.core import exceptions
 from django.db import connection
@@ -850,6 +849,13 @@ def delete(
             return collector.delete()
 
     def cached_clean_fields(self, fk_lookup_cache):
+        """
+        Immediately validates all fields, but uses a cache for foreign key (FK) lookups to reduce
+        repeated queries for many records with the same FK
+
+        :param fk_lookup_cache: A dictionary to use as a cache to prevent querying the database if a
+            FK exists in the cache, having already been validated
+        """
         excluded_fields = []
         fk_fields = [
             field for field in self._meta.fields if isinstance(field, models.ForeignKey)
@@ -883,10 +889,11 @@ def cached_clean_fields(self, fk_lookup_cache):
 
     def deferred_clean_fields(self):
         """
-        Calls `.clean_fields()` but excludes all foreign key fields and instead adds them to the
-        `fk_references` dictionary for deferred batch processing
+        Calls `.clean_fields()` but excludes all foreign key fields and instead returns them as a
+        dictionary for deferred batch processing
 
-        :param fk_references: A dictionary passed by reference
+        :return: A dictionary containing lists of `ForeignKeyReference`s keyed by the name of the
+            model being referenced by the FK
         """
         excluded_fields = []
         deferred_fks = defaultdict(list)

morango/sync/backends/base.py

@@ -21,6 +21,22 @@ def _create_placeholder_list(self, fields, db_values):
     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}

morango/sync/backends/postgres.py

@@ -52,7 +52,8 @@ def _prepare_set_casted_values(self, fields, source_table):
     def _bulk_full_record_upsert(self, cursor, table_name, fields, db_values):
         pk = get_pk_field(fields)
 
-        insert = """
+        cte_name = "new_values"
+        upsert = """
             {cte},
             updated as
             (
@@ -66,21 +67,18 @@ def _bulk_full_record_upsert(self, cursor, table_name, fields, db_values):
             SELECT {select_fields}
             FROM {cte_name} cte
             WHERE cte.{pk_field}::{pk_type} NOT IN (SELECT {pk_field} FROM updated)
-        """
-
-        cte_name = "new_values"
-        insert = insert.format(
+        """.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_name),
+            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(insert, db_values)
+        cursor.execute(upsert, db_values)
 
     def _bulk_update(self, cursor, table_name, fields, db_values):
         pk = get_pk_field(fields)
@@ -99,7 +97,7 @@ def _bulk_update(self, cursor, table_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_name),
+            set_values=self._prepare_set_casted_values(fields, "cte"),
             pk_field=pk.column,
             pk_type=pk.rel_db_type(self.connection),
         )

morango/sync/backends/sqlite.py

@@ -29,17 +29,32 @@ def _bulk_full_record_upsert(self, cursor, table_name, fields, db_values):
             for x in range(0, len(placeholder_list), num_of_rows_able_to_insert)
         ]
         # insert data chunks
-        fields = str(tuple(str(f.attname) for f in fields)).replace("'", "")
+        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}
+            insert = """
+                REPLACE INTO {table_name} {fields}
+                VALUES {placeholder_str}
             """.format(
-                table_name=table_name, fields=fields, placeholder_str=placeholder_str
+                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:
@@ -90,7 +105,6 @@ def _bulk_update(self, cursor, table_name, fields, db_values):
                 ),
             )
             # use DB-APIs parameter substitution (2nd parameter expects a sequence)
-            print("EXECUTE BULK UPDATE:", update, params)
             cursor.execute(update, params)
 
     def _dequeuing_merge_conflict_rmcb(self, cursor, transfersession_id):

morango/sync/backends/utils.py

@@ -67,6 +67,12 @@ class TemporaryTable(object):
     __slots__ = ("connection", "name", "fields", "backend", "_meta")
 
     def __init__(self, connection, name, **fields):
+        """
+        :param connection: A database connection object
+        :param name: A str name for the table in the database
+        :param fields: Keyword arguments are assumed to be fields for defining the schema of the
+            temporary table
+        """
         self.connection = connection
         self.name = name
         self.fields = []
@@ -86,13 +92,27 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
     @property
     def sql_name(self):
+        """
+        :return: The name actual name of the table used in the DB, prefixed to avoid collisions
+        """
         return self.connection.ops.quote_name("t_{}".format(self.name))
 
+    def get_field(self, name):
+        """
+        :param name: A str of the name of which field to find
+        :return: The field object
+        """
+        return next(f for f in self.fields if f.name == name)
+
     def create(self):
+        """
+        Creates the temporary table within the database
+        """
         fields = []
         params = []
         with self.connection.schema_editor() as schema_editor:
             for field in self.fields:
+                # generates the SQL expression for the table column
                 field_sql, field_params = schema_editor.column_sql(
                     self, field, include_default=True
                 )
@@ -103,16 +123,24 @@ def create(self):
             self.backend._create_temporary_table(c, self.sql_name, fields, params)
 
     def drop(self):
+        """
+        Drops the temporary table within the database
+        """
         with self.connection.cursor() as c:
             c.execute("DROP TABLE IF EXISTS {name}".format(name=self.sql_name))
 
     def bulk_insert(self, values):
+        """
+        Bulk inserts a list of records into the temporary table
+
+        :param values: A list of dictionaries containing data to insert, keyed by field name
+        """
         params = []
         for value_dict in values:
             for field in self.fields:
                 params.append(value_dict.get(field.attname))
         with self.connection.cursor() as c:
-            self.backend._bulk_full_record_upsert(c, self.sql_name, self.fields, params)
+            self.backend._bulk_insert(c, self.sql_name, self.fields, params)
 
     class Meta:
         """