Scoped field updates and context initialization issue

CHANGELOG.md

@@ -4,6 +4,8 @@ List of the most important changes for each release.
 
 ## 0.8.15
 - Improves buffer serialization performance by performing bulk counter lookup
+- Corrects an initialization issue with `SessionContext` that leads to an incorrect value for `is_push`
+- Save calls to Morango models during a sync are now scoped to only the changed fields
 
 ## 0.8.14
 - Adds utility for addressing immediate FK constraints caused by Django upgrade, automatically performed for morango models in a Django migration.

morango/models/core.py

@@ -319,6 +319,7 @@ def update_state(self, stage=None, stage_status=None):
         :type stage: morango.constants.transfer_stages.*|None
         :type stage_status: morango.constants.transfer_statuses.*|None
         """
+        update_fields = []
         if stage is not None:
             if self.transfer_stage and transfer_stages.stage(
                 self.transfer_stage
@@ -329,13 +330,18 @@ def update_state(self, stage=None, stage_status=None):
                     )
                 )
             self.transfer_stage = stage
+            update_fields.append("transfer_stage")
+
         if stage_status is not None:
             self.transfer_stage_status = stage_status
-        if stage is not None or stage_status is not None:
+            update_fields.append("transfer_stage_status")
+
+        if update_fields:
             self.last_activity_timestamp = timezone.now()
-            self.save()
+            update_fields.append("last_activity_timestamp")
+            self.save(update_fields=update_fields)
             self.sync_session.last_activity_timestamp = timezone.now()
-            self.sync_session.save()
+            self.sync_session.save(update_fields=["last_activity_timestamp"])
 
     def delete_buffers(self):
         """

morango/sync/context.py

@@ -52,7 +52,7 @@ def __init__(
 
         if self.transfer_session:
             self.sync_session = transfer_session.sync_session or self.sync_session
-            self.is_push = transfer_session.push or self.is_push
+            self.is_push = transfer_session.push
             if transfer_session.filter:
                 self.filter = transfer_session.get_filter()
 

morango/sync/operations.py

@@ -1107,9 +1107,11 @@ def handle(self, context):
             context.transfer_session.client_fsic = context.request.data.get(
                 "client_fsic", "{}"
             )
+            update_fields = ["server_fsic", "client_fsic"]
         else:
             context.transfer_session.client_fsic = fsic
-        context.transfer_session.save()
+            update_fields = ["client_fsic"]
+        context.transfer_session.save(update_fields=update_fields)
         return transfer_statuses.COMPLETED
 
 
@@ -1138,7 +1140,7 @@ def handle(self, context):
 
         logger.debug("[morango] Queued {} records".format(records_total))
         context.transfer_session.records_total = records_total
-        context.transfer_session.save()
+        context.transfer_session.save(update_fields=["records_total"])
         return transfer_statuses.COMPLETED
 
 
@@ -1342,7 +1344,7 @@ def handle(self, context):
             context.transfer_session.delete_buffers()
 
         context.transfer_session.active = False
-        context.transfer_session.save()
+        context.transfer_session.save(update_fields=["active"])
         return transfer_statuses.COMPLETED
 
 
@@ -1493,15 +1495,17 @@ def handle(self, context):
 
         data = self.create_transfer_session(context)
         context.transfer_session.server_fsic = data.get("server_fsic") or "{}"
+        update_fields = ["server_fsic"]
 
         # A legacy instance performs queuing during the creation of the transfer session, so since we use a new
         # workflow we need to update the network server when pushing to say how many records we've queued. For pull,
         # we handle that here in the initialization/creation of the transfer session,
         # since that's when it's first available.
         if context.transfer_session.pull:
             context.transfer_session.records_total = data.get("records_total", 0)
+            update_fields.append("records_total")
 
-        context.transfer_session.save()
+        context.transfer_session.save(update_fields=update_fields)
         return transfer_statuses.COMPLETED
 
 
@@ -1569,7 +1573,7 @@ def handle(self, context):
 
         if remote_status == transfer_statuses.COMPLETED:
             context.transfer_session.server_fsic = data.get("server_fsic")
-            context.transfer_session.save()
+            context.transfer_session.save(update_fields=["server_fsic"])
 
         return remote_status
 
@@ -1618,7 +1622,7 @@ def handle(self, context):
 
         if context.is_pull and remote_status == transfer_statuses.COMPLETED:
             context.transfer_session.records_total = data.get("records_total", 0)
-            context.transfer_session.save()
+            context.transfer_session.save(update_fields=["records_total"])
 
         return remote_status
 
@@ -1654,7 +1658,7 @@ def handle(self, context):
         )
         context.transfer_session.bytes_sent = context.connection.bytes_sent
         context.transfer_session.bytes_received = context.connection.bytes_received
-        context.transfer_session.save()
+        context.transfer_session.save(update_fields=["records_transferred", "bytes_sent", "bytes_received"])
 
         # if we've transferred all records, return a completed status
         op_status = transfer_statuses.PENDING

morango/sync/utils.py

@@ -137,16 +137,17 @@ def validate_and_create_buffer_data(  # noqa: C901
             model_uuid__in=[record["model_uuid"] for record in data]
         ).delete()
 
+        update_fields = ["records_transferred"]
         if connection is not None:
             transfer_session.bytes_sent = connection.bytes_sent
-        if connection is not None:
             transfer_session.bytes_received = connection.bytes_received
+            update_fields.extend(["bytes_sent", "bytes_received"])
 
         Buffer.objects.bulk_create(buffer_list)
         RecordMaxCounterBuffer.objects.bulk_create(rmcb_list)
 
         transfer_session.records_transferred += len(buffer_list) - deleted_buffers
-        transfer_session.save()
+        transfer_session.save(update_fields=update_fields)
 
 
 class SyncSignal(object):

tests/testapp/tests/sync/test_context.py

@@ -67,6 +67,24 @@ def test_init__with_transfer_session(self):
         self.assertFalse(context.is_push)
         self.assertTrue(context.is_pull)
 
+    def test_init__with_transfer_session__direction_handling(self):
+        sync_session = mock.Mock(spec=SyncSession)
+        sync_filter = Filter("before_filter")
+        transfer_session = mock.Mock(
+            spec=TransferSession,
+            sync_session=sync_session,
+            push=False,
+            filter="after_filter",
+            transfer_stage=transfer_stages.TRANSFERRING,
+            transfer_stage_status=transfer_statuses.STARTED,
+        )
+        transfer_session.get_filter.return_value = Filter(transfer_session.filter)
+
+        # the transfer session's `push=False` should take precedence over the constructor arg
+        context = TestSessionContext(transfer_session=transfer_session, sync_filter=sync_filter, is_push=True)
+        self.assertFalse(context.is_push)
+        self.assertTrue(context.is_pull)
+
     def test_init__with_transfer_session__no_filter(self):
         sync_session = mock.Mock(spec=SyncSession)
         sync_filter = Filter("before_filter")