Index: backend/tests/baserow/contrib/database/api/fields/dependencies/test_update_collector.py
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/backend/tests/baserow/contrib/database/api/fields/dependencies/test_update_collector.py b/backend/tests/baserow/contrib/database/api/fields/dependencies/test_update_collector.py
--- a/backend/tests/baserow/contrib/database/api/fields/dependencies/test_update_collector.py	(revision 40014ab4eaf5c39bffd637a5784b772a3e122784)
+++ b/backend/tests/baserow/contrib/database/api/fields/dependencies/test_update_collector.py	(date 1655712496254)
@@ -287,3 +287,88 @@
     assert send_mock.call_args[1]["field"].id == first_table_primary_field.id
     assert send_mock.call_args[1]["user"] is None
     assert send_mock.call_args[1]["related_fields"] == [first_table_other_field]
+
+
+@pytest.mark.django_db
+@patch("baserow.contrib.database.fields.signals.field_updated.send")
+def test_can_update_rows_joined_to_a_starting_row_across_a_m2m_using_python(
+    send_mock, api_client, data_fixture, django_assert_num_queries
+):
+    user = data_fixture.create_user()
+    database = data_fixture.create_database_application(user=user)
+    first_table = data_fixture.create_database_table(database=database)
+    second_table = data_fixture.create_database_table(database=database)
+    first_table_primary_field = data_fixture.create_text_field(
+        name="primary", primary=True, table=first_table
+    )
+    first_table_text_field = data_fixture.create_text_field(
+        name="text", table=first_table
+    )
+    data_fixture.create_text_field(name="primary", primary=True, table=second_table)
+    link_row_field = FieldHandler().create_field(
+        user=user,
+        table=first_table,
+        type_name="link_row",
+        link_row_table=second_table,
+        name="link",
+    )
+    first_table_model = first_table.get_model(attribute_names=True)
+    second_table_model = second_table.get_model(attribute_names=True)
+
+    second_table_a_row = second_table_model.objects.create(primary="a")
+    second_table_b_row = second_table_model.objects.create(primary="b")
+
+    first_table_1_row = first_table_model.objects.create(primary="1")
+    first_table_2_row = first_table_model.objects.create(primary="2")
+
+    first_table_1_row.link.add(second_table_a_row.id)
+    first_table_1_row.link.add(second_table_b_row.id)
+    first_table_1_row.save()
+
+    first_table_2_row.link.add(second_table_b_row.id)
+    first_table_2_row.save()
+
+    update_collector = CachingFieldUpdateCollector(
+        second_table, starting_row_id=second_table_b_row.id
+    )
+
+    def update_func1(rows):
+        for row in rows:
+            setattr(row, first_table_primary_field.db_column, f"row {row.id}")
+
+    def update_func2(rows):
+        for row in rows:
+            setattr(row, first_table_text_field.db_column, f"func2 row {row.id}")
+
+    update_collector.add_field_with_pending_update_function(
+        first_table_primary_field,
+        update_func1,
+        via_path_to_starting_table=[link_row_field],
+    )
+    update_collector.add_field_with_pending_update_function(
+        first_table_text_field,
+        update_func2,
+        via_path_to_starting_table=[link_row_field],
+    )
+    # Cache the models so we are only asserting about the update queries
+    update_collector.cache_model(first_table.get_model())
+    update_collector.cache_model(second_table.get_model())
+    # One query to get the rows for the update funcs, the second being a bulk update.
+    with django_assert_num_queries(2):
+        updated_fields = update_collector.apply_updates_and_get_updated_fields()
+
+    # No field in the starting table (second_table) was updated
+    assert updated_fields == []
+    first_table_1_row.refresh_from_db()
+    first_table_2_row.refresh_from_db()
+    assert first_table_1_row.primary == f"row {first_table_1_row.id}"
+    assert first_table_1_row.text == f"func2 row {first_table_1_row.id}"
+    assert first_table_2_row.primary == f"row {first_table_2_row.id}"
+    assert first_table_2_row.text == f"func2 row {first_table_2_row.id}"
+
+    send_mock.assert_not_called()
+    update_collector.send_additional_field_updated_signals()
+    send_mock.assert_called_once()
+    assert send_mock.call_args[1]["field"].id == first_table_primary_field.id
+    assert send_mock.call_args[1]["user"] is None
+    assert send_mock.call_args[1]["related_fields"] == [first_table_text_field]
Index: backend/src/baserow/contrib/database/fields/dependencies/update_collector.py
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/backend/src/baserow/contrib/database/fields/dependencies/update_collector.py b/backend/src/baserow/contrib/database/fields/dependencies/update_collector.py
--- a/backend/src/baserow/contrib/database/fields/dependencies/update_collector.py	(revision 40014ab4eaf5c39bffd637a5784b772a3e122784)
+++ b/backend/src/baserow/contrib/database/fields/dependencies/update_collector.py	(date 1655712542862)
@@ -1,5 +1,5 @@
 from collections import defaultdict
-from typing import Optional, Dict, List, Tuple
+from typing import Optional, Dict, List, Tuple, Callable
 
 from django.db.models import Expression
 
@@ -13,6 +13,9 @@
 class PathBasedUpdateStatementCollector:
     def __init__(self, table: Table, field_cache: FieldCache):
         self.update_statements: Dict[str, Expression] = {}
+        self.python_cell_update_functions: Dict[
+            str, Callable[[List[GeneratedTableModel]], None]
+        ] = {}
         self.table = table
         self.sub_paths: Dict[str, PathBasedUpdateStatementCollector] = {}
         self.field_cache = field_cache
@@ -28,18 +31,38 @@
                 raise InvalidViaPath()
             self.update_statements[field.db_column] = update_statement
         else:
-            next_via_field_link = path_from_starting_table[0]
-            if next_via_field_link.link_row_table != self.table:
-                raise InvalidViaPath()
-            next_link_db_column = next_via_field_link.db_column
-            if next_link_db_column not in self.sub_paths:
-                self.sub_paths[next_link_db_column] = PathBasedUpdateStatementCollector(
-                    next_via_field_link.table,
-                    self.field_cache,
-                )
-            self.sub_paths[next_link_db_column].add_update_statement(
-                field, update_statement, path_from_starting_table[1:]
+            sub_path = self._get_or_create_sub_path(path_from_starting_table)
+            sub_path.add_update_statement(
+                field, update_statement, path_from_starting_table[1:]
+            )
+
+    def add_python_cell_update_function(
+        self,
+        field: Field,
+        func: Callable[[List[GeneratedTableModel]], None],
+        path_from_starting_table: Optional[List[LinkRowField]] = None,
+    ):
+        if not path_from_starting_table:
+            if self.table != field.table:
+                raise InvalidViaPath()
+            self.python_cell_update_functions[field.db_column] = func
+        else:
+            sub_path = self._get_or_create_sub_path(path_from_starting_table)
+            sub_path.add_python_cell_update_function(
+                field, func, path_from_starting_table[1:]
+            )
+
+    def _get_or_create_sub_path(self, path_from_starting_table):
+        next_via_field_link = path_from_starting_table[0]
+        if next_via_field_link.link_row_table != self.table:
+            raise InvalidViaPath()
+        next_link_db_column = next_via_field_link.db_column
+        if next_link_db_column not in self.sub_paths:
+            self.sub_paths[next_link_db_column] = PathBasedUpdateStatementCollector(
+                next_via_field_link.table,
+                self.field_cache,
             )
+        return self.sub_paths[next_link_db_column]
 
     def execute_all(
         self,
@@ -69,8 +92,29 @@
             if isinstance(starting_row_id, list):
                 path_to_starting_table_id_column += "__in"
             qs = qs.filter(**{path_to_starting_table_id_column: starting_row_id})
+
+        if self.python_cell_update_functions:
+            rows_to_bulk_update = [row for row in qs]
+            for field_name, func in self.python_cell_update_functions.items():
+                func(rows_to_bulk_update)
+            model.objects_and_trash.bulk_update(
+                rows_to_bulk_update, self.python_cell_update_functions.keys()
+            )
+
         qs.update(**self.update_statements)
 
+    def _get_or_create_child_collector(self, path_from_starting_table):
+        next_via_field_link = path_from_starting_table[0]
+        if next_via_field_link.link_row_table != self.table:
+            raise InvalidViaPath()
+        next_link_db_column = next_via_field_link.db_column
+        if next_link_db_column not in self.sub_paths:
+            self.sub_paths[next_link_db_column] = PathBasedUpdateStatementCollector(
+                next_via_field_link.table,
+                self.field_cache,
+            )
+        return next_link_db_column
+
 
 class CachingFieldUpdateCollector(FieldCache):
     """
@@ -134,6 +178,38 @@
             field, update_statement, via_path_to_starting_table
         )
 
+    def add_field_with_pending_update_function(
+        self,
+        field: Field,
+        update_function: Callable[[List[GeneratedTableModel]], None],
+        via_path_to_starting_table: Optional[List[LinkRowField]] = None,
+    ):
+        """
+        Stores the provided field as an updated one to send in field updated signals
+        when triggered to do so. Also stores the provided update function which will
+        be called with the list of rows that are being updated containing the provided
+        field.
+
+        :param field: The field which has been updated.
+        :param update_function: An update function that will be called with the rows
+            that need to be updated as their field cell values should be changed.
+            The function should for each row in rows `setattr(row, field.db_column,
+            your_new_value)` as the row instances will be passed to a Django bulk_update
+            call after the function has returned and so your func should directly modify
+            the row attributes to get changes saved to the db.
+        :param via_path_to_starting_table: A list of link row fields which lead from
+            the self.starting_table to the table containing field. Used to properly
+            order the update statements so the graph is updated in sequence and also
+            used if self.starting_row_id is set so only rows which join back to the
+            starting row via this path are updated.
+        """
+
+        # noinspection PyTypeChecker
+        self._updated_fields_per_table[field.table_id][field.id] = field
+        self._update_statement_collector.add_python_cell_update_function(
+            field, update_function, via_path_to_starting_table
+        )
+
     def apply_updates_and_get_updated_fields(self) -> List[Field]:
         """
         Triggers all update statements to be executed in the correct order in as few
