Add associated document id field to ChangeMeta

corehq/ex-submodules/pillowtop/feed/interface.py

@@ -40,6 +40,9 @@ class ChangeMeta(jsonobject.JsonObject):
     # track when first published (will not get updated on retry, unlike publish_timestamp)
     original_publication_datetime = jsonobject.DateTimeProperty(default=datetime.utcnow)
 
+    # available to hold any associated document. For cases, this is the form ID responsible for the change
+    associated_document_id = jsonobject.StringProperty()
+
 
 class Change(object):
     """
@@ -119,10 +122,10 @@ def __getitem__(self, key):
         return self._dict[key]
 
     def __setitem__(self, key, value):
-        raise NotImplemented('This is a read-only dictionary!')
+        raise NotImplementedError('This is a read-only dictionary!')
 
     def __delitem__(self, key, value):
-        raise NotImplemented('This is a read-only dictionary!')
+        raise NotImplementedError('This is a read-only dictionary!')
 
     def __iter__(self):
         return iter(self._dict)
@@ -134,7 +137,7 @@ def get(self, key, default=None):
         return self._dict.get(key, default)
 
     def pop(self, key, default):
-        raise NotImplemented('This is a read-only dictionary!')
+        raise NotImplementedError('This is a read-only dictionary!')
 
     def to_dict(self):
         return self._dict

corehq/form_processor/backends/sql/processor.py

@@ -145,7 +145,11 @@ def publish_changes_to_kafka(processed_forms, cases, stock_result):
         publish_form_saved(processed_forms.submitted)
         cases = cases or []
         for case in cases:
-            publish_case_saved(case, send_post_save_signal=False)
+            publish_case_saved(
+                case,
+                associated_form_id=processed_forms.submitted.form_id,
+                send_post_save_signal=False
+            )
 
         if stock_result:
             for ledger in stock_result.models_to_save:

corehq/form_processor/change_publishers.py

@@ -31,16 +31,16 @@ def publish_form_deleted(domain, form_id):
     ))
 
 
-def publish_case_saved(case, send_post_save_signal=True):
+def publish_case_saved(case, associated_form_id=None, send_post_save_signal=True):
     """
     Publish the change to kafka and run case post-save signals.
     """
-    producer.send_change(topics.CASE_SQL, change_meta_from_sql_case(case))
+    producer.send_change(topics.CASE_SQL, change_meta_from_sql_case(case, associated_form_id))
     if send_post_save_signal:
         sql_case_post_save.send(case.__class__, case=case)
 
 
-def change_meta_from_sql_case(case):
+def change_meta_from_sql_case(case, associated_form_id=None):
     return ChangeMeta(
         document_id=case.case_id,
         data_source_type=data_sources.SOURCE_SQL,
@@ -49,6 +49,7 @@ def change_meta_from_sql_case(case):
         document_subtype=case.type,
         domain=case.domain,
         is_deletion=case.is_deleted,
+        associated_document_id=associated_form_id
     )