Merge pull request #35054 from dimagi/sk/expression-repeater-response

Use the HTTP response from a Configurable Repeater to update a case

corehq/apps/hqcase/api/updates.py

@@ -158,7 +158,7 @@ def get_case_id(self, case_db):
         return case_db.get_by_external_id(self.external_id)
 
 
-def handle_case_update(domain, data, user, device_id, is_creation):
+def handle_case_update(domain, data, user, device_id, is_creation, xmlns=None):
     is_bulk = isinstance(data, list)
     if is_bulk:
         updates = _get_bulk_updates(domain, data, user)
@@ -171,7 +171,7 @@ def handle_case_update(domain, data, user, device_id, is_creation):
         validate_update_permission(domain, updates, user, case_db)
 
     case_blocks = [update.get_caseblock(case_db) for update in updates]
-    xform, cases = _submit_case_blocks(case_blocks, domain, user, device_id)
+    xform, cases = _submit_case_blocks(case_blocks, domain, user, device_id, xmlns)
     if xform.is_error:
         raise SubmissionError(xform.problem, xform.form_id,)
 
@@ -289,13 +289,13 @@ def _get_owner_id(self):
         }
 
 
-def _submit_case_blocks(case_blocks, domain, user, device_id):
+def _submit_case_blocks(case_blocks, domain, user, device_id, xmlns):
     return submit_case_blocks(
         case_blocks=case_blocks,
         domain=domain,
         username=user.username,
         user_id=user.user_id,
-        xmlns='http://commcarehq.org/case_api',
+        xmlns=xmlns or 'http://commcarehq.org/case_api',
         device_id=device_id,
         max_wait=15
     )

corehq/apps/hqcase/utils.py

@@ -26,13 +26,15 @@
 SYSTEM_FORM_XMLNS = 'http://commcarehq.org/case'
 EDIT_FORM_XMLNS = 'http://commcarehq.org/case/edit'
 AUTO_UPDATE_XMLNS = 'http://commcarehq.org/hq_case_update_rule'
+REPEATER_RESPONSE_XMLNS = 'http://commcarehq.org/data_forwarding/response'
 
 SYSTEM_FORM_XMLNS_MAP = {
     SYSTEM_FORM_XMLNS: gettext_lazy('System Form'),
     EDIT_FORM_XMLNS: gettext_lazy('Data Cleaning Form'),
     AUTO_UPDATE_XMLNS: gettext_lazy('Automatic Case Update Rule'),
     DEDUPE_XMLNS: gettext_lazy('Deduplication Rule'),
-    XMLNS_DHIS2: gettext_lazy('DHIS2 Integration')
+    XMLNS_DHIS2: gettext_lazy('DHIS2 Integration'),
+    REPEATER_RESPONSE_XMLNS: gettext_lazy('Data Forwarding Response'),
 }
 
 ALLOWED_CASE_IDENTIFIER_TYPES = [

corehq/apps/userreports/ui/fields.py

@@ -52,14 +52,20 @@ def to_python(self, value):
             return value
 
         val = super(JsonField, self).to_python(value)
+        if val in self.null_values:
+            return val
+
         try:
             return json.loads(val)
-        except:
+        except Exception:
             raise forms.ValidationError(_('Please enter valid JSON. This is not valid: {}'.format(value)))
 
     def validate(self, value):
-        if value in self.null_values and self.required:
-            raise forms.ValidationError(self.error_messages['required'])
+        if value in self.null_values:
+            if self.required:
+                raise forms.ValidationError(self.error_messages['required'])
+            return
+
         if self.expected_type and not isinstance(value, self.expected_type):
             raise forms.ValidationError(
                 _('Expected {} but was {}').format(self.expected_type.__name__, type(value).__name__))

corehq/form_processor/models/cases.py

@@ -1216,6 +1216,15 @@ def get_most_recent_form_transaction(self, case_id):
             .first()
         )
 
+    def get_last_n_recent_form_transaction(self, case_id, limit):
+        return (
+            self.partitioned_query(case_id)
+            .filter(case_id=case_id, revoked=False)
+            .annotate(type_filter=F('type').bitand(self.model.TYPE_FORM))
+            .filter(type_filter=self.model.TYPE_FORM)
+            .order_by("-server_date")[:limit]
+        )
+
     def get_transactions_by_type(self, case_id, transaction_type):
         return list(self.plproxy_raw(
             'SELECT * from get_case_transactions_by_type(%s, %s)',

corehq/motech/repeaters/expression/forms.py

@@ -10,6 +10,12 @@
 from corehq.motech.repeaters.forms import GenericRepeaterForm
 
 
+LABELS = {
+    'case_action_filter_expression': _("Response case action filter expression"),
+    'case_action_expression': _("Response case action expression"),
+}
+
+
 class BaseExpressionRepeaterForm(GenericRepeaterForm):
     configured_filter = JsonField(expected_type=dict, help_text=help_text.CONFIGURED_FILTER)
     configured_expression = JsonField(expected_type=dict)
@@ -18,9 +24,34 @@ class BaseExpressionRepeaterForm(GenericRepeaterForm):
         help_text=_("Items to add to the end of the URL. Please see the documentation for more information.")
     )
 
+    case_action_filter_expression = JsonField(
+        expected_type=dict, required=False,
+        label=LABELS['case_action_filter_expression'],
+        help_text=_(
+            "Use this to determine if the response should create or update a case. "
+            "If left blank, no action will be taken. "
+            'For more info see <a target="_blank" href="'
+            'https://dimagi.atlassian.net/wiki/spaces/GS/pages/2146602964/Configurable+Repeaters'
+            '">these docs</a>'
+        )
+    )
+    case_action_expression = JsonField(
+        expected_type=dict, required=False,
+        label=LABELS['case_action_expression'],
+        help_text=_(
+            "Use this to create a Case API payload which will be used to create or update a case. "
+            'For more info see <a target="_blank" href="'
+            'https://dimagi.atlassian.net/wiki/spaces/GS/pages/2146602964/Configurable+Repeaters'
+            '">these docs</a>'
+        )
+    )
+
     def get_ordered_crispy_form_fields(self):
         fields = super().get_ordered_crispy_form_fields()
-        return fields + ['url_template', 'configured_filter', 'configured_expression']
+        return fields + [
+            'url_template', 'configured_filter', 'configured_expression',
+            'case_action_filter_expression', 'case_action_expression'
+        ]
 
     def clean_configured_expression(self):
         try:
@@ -39,3 +70,39 @@ def clean_configured_filter(self):
             raise ValidationError(e)
 
         return self.cleaned_data['configured_filter']
+
+    def clean_case_action_expression(self):
+        raw = self.cleaned_data.get('case_action_expression')
+        if raw:
+            try:
+                ExpressionFactory.from_spec(
+                    raw, FactoryContext.empty(domain=self.domain)
+                )
+            except BadSpecError as e:
+                raise ValidationError(e)
+
+        return raw
+
+    def clean_case_action_filter_expression(self):
+        raw = self.cleaned_data.get('case_action_filter_expression')
+        if raw:
+            try:
+                FilterFactory.from_spec(raw)
+            except BadSpecError as e:
+                raise ValidationError(e)
+
+        return raw
+
+    def clean(self):
+        cleaned_data = super().clean()
+        case_filter = bool(cleaned_data.get('case_action_filter_expression'))
+        case_operation = bool(cleaned_data.get('case_action_expression'))
+        if case_filter ^ case_operation:
+            field = 'case_action_expression' if case_filter else 'case_action_filter_expression'
+            other = 'case_action_filter_expression' if case_filter else 'case_action_expression'
+            raise ValidationError({
+                field: _("This field is required when '{other}' is provided").format(
+                    other=LABELS[other]
+                ),
+            })
+        return cleaned_data

corehq/motech/repeaters/expression/repeaters.py

@@ -1,20 +1,29 @@
-from django.utils.translation import gettext_lazy as _
+import logging
+from json import JSONDecodeError
 
+from django.utils.translation import gettext_lazy as _
 from memoized import memoized
 
+from corehq.apps.hqcase.api.updates import handle_case_update
+from corehq.apps.hqcase.case_helper import UserDuck
+from corehq.apps.hqcase.utils import REPEATER_RESPONSE_XMLNS
 from corehq.apps.userreports.expressions.factory import ExpressionFactory
 from corehq.apps.userreports.filters.factory import FilterFactory
 from corehq.apps.userreports.specs import EvaluationContext, FactoryContext
-from corehq.form_processor.models import CommCareCase, XFormInstance
-from corehq.motech.repeaters.expression.repeater_generators import (
-    ExpressionPayloadGenerator,
-)
-from corehq.motech.repeaters.models import OptionValue, Repeater
+from corehq.form_processor.models import CaseTransaction, CommCareCase, XFormInstance
 from corehq.motech.repeaters.expression.repeater_generators import (
     ArcGISFormExpressionPayloadGenerator,
     FormExpressionPayloadGenerator,
 )
-from corehq.toggles import EXPRESSION_REPEATER, ARCGIS_INTEGRATION
+from corehq.motech.repeaters.expression.repeater_generators import (
+    ExpressionPayloadGenerator,
+)
+from corehq.motech.repeaters.models import OptionValue, Repeater
+from corehq.motech.repeaters.models import is_response, is_success_response
+from corehq.toggles import ARCGIS_INTEGRATION, EXPRESSION_REPEATER
+from dimagi.utils.logging import notify_exception
+
+logger = logging.getLogger(__name__)
 
 
 class BaseExpressionRepeater(Repeater):
@@ -28,6 +37,9 @@ class Meta:
     configured_expression = OptionValue(default=dict)
     url_template = OptionValue(default=None)
 
+    case_action_filter_expression = OptionValue(default=dict)
+    case_action_expression = OptionValue(default=dict)
+
     payload_generator_classes = (ExpressionPayloadGenerator,)
 
     @property
@@ -40,6 +52,20 @@ def parsed_filter(self):
     def parsed_expression(self):
         return ExpressionFactory.from_spec(self.configured_expression, FactoryContext.empty(domain=self.domain))
 
+    @property
+    @memoized
+    def parsed_case_action_filter(self):
+        return FilterFactory.from_spec(
+            self.case_action_filter_expression, FactoryContext.empty(domain=self.domain)
+        )
+
+    @property
+    @memoized
+    def parsed_case_action_expression(self):
+        return ExpressionFactory.from_spec(
+            self.case_action_expression, FactoryContext.empty(domain=self.domain)
+        )
+
     @classmethod
     def available_for_domain(cls, domain):
         return EXPRESSION_REPEATER.enabled(domain)
@@ -66,6 +92,40 @@ def get_url(self, repeat_record):
             )
         return base_url
 
+    def handle_response(self, response, repeat_record):
+        super().handle_response(response, repeat_record)
+        if self.case_action_filter_expression and is_response(response):
+            try:
+                self._process_response_as_case_update(response, repeat_record)
+            except Exception as e:
+                notify_exception(None, "Error processing response from Repeater request", e)
+
+    def _process_response_as_case_update(self, response, repeat_record):
+        domain = repeat_record.domain
+        context = get_evaluation_context(domain, repeat_record, self.payload_doc(repeat_record), response)
+        if not self.parsed_case_action_filter(context.root_doc, context):
+            return False
+
+        self._perform_case_update(domain, context)
+        return True
+
+    def _perform_case_update(self, domain, context):
+        data = self.parsed_case_action_expression(context.root_doc, context)
+        if data:
+            data = data if isinstance(data, list) else [data]
+            handle_case_update(
+                domain=domain,
+                data=data,
+                user=UserDuck('system', ''),
+                device_id=self.device_id,
+                is_creation=False,
+                xmlns=REPEATER_RESPONSE_XMLNS,
+            )
+
+    @property
+    def device_id(self):
+        return f'{__name__}.{self.__class__.__name__}:{self.id}'
+
 
 class CaseExpressionRepeater(BaseExpressionRepeater):
 
@@ -83,6 +143,26 @@ def form_class_name(self):
     def payload_doc(self, repeat_record):
         return CommCareCase.objects.get_case(repeat_record.payload_id, repeat_record.domain).to_json()
 
+    def allowed_to_forward(self, payload):
+        allowed = super().allowed_to_forward(payload)
+        if allowed:
+            transactions = CaseTransaction.objects.get_last_n_recent_form_transaction(payload.case_id, 2)
+            # last 2 transactions were from repeater updates. This suggests a cycle.
+            possible_cycle = {t.xmlns for t in transactions} == {REPEATER_RESPONSE_XMLNS}
+            if possible_cycle:
+                logger.warning(
+                    f"Possible data forwarding loop detected for case {payload.case_id}. "
+                    f"Transactions: {[t.id for t in transactions]}"
+                )
+                return False
+            last_transaction = transactions[0] if transactions else None
+            return last_transaction and not (
+                # last update was from this repeater, ignore
+                last_transaction.xmlns == REPEATER_RESPONSE_XMLNS
+                and last_transaction.device_id == self.device_id
+            )
+        return False
+
 
 class FormExpressionRepeater(BaseExpressionRepeater):
 
@@ -124,3 +204,23 @@ def available_for_domain(cls, domain):
             super(ArcGISFormExpressionRepeater, cls).available_for_domain(domain)
             and ARCGIS_INTEGRATION.enabled(domain)
         )
+
+
+def get_evaluation_context(domain, repeat_record, payload_doc, response):
+    try:
+        body = response.json()
+    except JSONDecodeError:
+        body = response.text
+    return EvaluationContext({
+        'domain': domain,
+        'success': is_success_response(response),
+        'payload': {
+            'id': repeat_record.payload_id,
+            'doc': payload_doc,
+        },
+        'response': {
+            'status_code': response.status_code,
+            'headers': response.headers,
+            'body': body,
+        },
+    })