Merge pull request #1108 from hinashi/feature/type_checking

Added type checking to trigger model

Author: userlocalhost

trigger/models.py

@@ -1,3 +1,5 @@
+from typing import TYPE_CHECKING
+
 from django.db import models
 
 from acl.models import ACLBase
@@ -8,6 +10,9 @@
 from entry.api_v2.serializers import EntryUpdateSerializer
 from entry.models import Attribute, Entry
 
+if TYPE_CHECKING:
+    from django.db.models import Manager
+
 
 ## These are internal classes for AirOne trigger and action
 class InputTriggerCondition(object):
@@ -158,82 +163,13 @@ def _do_get_value(input_value, attr_type):
         return _do_get_value(raw_input_value, self.attr.type)
 
 
-class TriggerAction(models.Model):
-    condition = models.ForeignKey("TriggerParent", on_delete=models.CASCADE, related_name="actions")
-    attr = models.ForeignKey("entity.EntityAttr", on_delete=models.CASCADE)
-
-    def save_actions(self, input: InputTriggerAction):
-        for input_action_value in input.values:
-            params = {
-                "action": self,
-                "str_cond": input_action_value.str_cond,
-                "ref_cond": input_action_value.ref_cond,
-                "bool_cond": input_action_value.bool_cond,
-            }
-            TriggerActionValue.objects.create(**params)
-
-    def get_serializer_acceptable_value(self, value=None, attr_type=None):
-        """
-        This converts TriggerActionValue to the value that EntryUpdateSerializer can accept.
-        """
-        if not attr_type:
-            attr_type = self.attr.type
-
-        value = value or self.values.first()
-        if attr_type & AttrTypeValue["array"]:
-            return [
-                self.get_serializer_acceptable_value(x, attr_type ^ AttrTypeValue["array"])
-                for x in self.values.all()
-            ]
-        elif attr_type == AttrTypeValue["boolean"]:
-            return value.bool_cond
-        elif attr_type == AttrTypeValue["named_object"]:
-            value.ref_cond.id if isinstance(value.ref_cond, Entry) else None
-            return {
-                "name": value.str_cond,
-                "id": value.ref_cond.id,
-            }
-        elif attr_type == AttrTypeValue["string"]:
-            return value.str_cond
-        elif attr_type == AttrTypeValue["text"]:
-            return value.str_cond
-        elif attr_type == AttrTypeValue["object"]:
-            return value.ref_cond.id if isinstance(value.ref_cond, Entry) else None
-
-    def run(self, user, entry, call_stacks=[]):
-        # When self.id contains in call_stacks, it means that this action is already invoked.
-        # This prevents infinite loop.
-        if self.id in call_stacks:
-            return
-
-        # update specified Entry by configured attribute value
-        setting_data = {
-            "id": entry.id,
-            "name": entry.name,
-            "attrs": [{"id": self.attr.id, "value": self.get_serializer_acceptable_value()}],
-            "delay_trigger": False,
-            "call_stacks": [*call_stacks, self.id],
-        }
-        serializer = EntryUpdateSerializer(
-            instance=entry, data=setting_data, context={"request": DRFRequest(user)}
-        )
-        if serializer:
-            serializer.is_valid(raise_exception=True)
-            serializer.save()
-
-
-class TriggerActionValue(models.Model):
-    action = models.ForeignKey(TriggerAction, on_delete=models.CASCADE, related_name="values")
-    str_cond = models.TextField(blank=True, null=True)
-    ref_cond = models.ForeignKey("entry.Entry", on_delete=models.SET_NULL, null=True, blank=True)
-    bool_cond = models.BooleanField(default=False)
-
-    # TODO: Add method to register value to Attribute when action is invoked
-
-
 class TriggerParent(models.Model):
     entity = models.ForeignKey("entity.Entity", on_delete=models.CASCADE)
 
+    if TYPE_CHECKING:
+        conditions: Manager["TriggerCondition"]
+        actions: Manager["TriggerAction"]
+
     def is_match_condition(self, inputs: list[InputTriggerCondition]):
         if all([c.is_same_condition(inputs) for c in self.conditions.all()]):
             return True
@@ -251,7 +187,7 @@ def save_conditions(self, inputs: list[InputTriggerCondition]):
             if not TriggerCondition.objects.filter(**params).exists():
                 TriggerCondition.objects.create(**params)
 
-    def get_actions(self, recv_attrs: list) -> list[TriggerAction]:
+    def get_actions(self, recv_attrs: list) -> list["TriggerAction"]:
         """
         This method checks whether specified entity's Trigger is invoked by recv_attrs context.
         The recv_attrs format should be compatible with APIv2 standard.
@@ -261,7 +197,7 @@ def get_actions(self, recv_attrs: list) -> list[TriggerAction]:
         context to reduce DB query to get it from Attribute instance.
         """
 
-        def _is_match(condition):
+        def _is_match(condition: TriggerCondition):
             for attr_info in [x for x in recv_attrs if x["attr_id"] == condition.attr.id]:
                 if condition.is_match_condition(attr_info["value"]):
                     return True
@@ -446,3 +382,79 @@ def get_invoked_actions(cls, entity: Entity, recv_data: list):
             actions += parent_condition.get_actions(params)
 
         return actions
+
+
+class TriggerAction(models.Model):
+    condition = models.ForeignKey(TriggerParent, on_delete=models.CASCADE, related_name="actions")
+    attr = models.ForeignKey("entity.EntityAttr", on_delete=models.CASCADE)
+
+    if TYPE_CHECKING:
+        values: Manager["TriggerActionValue"]
+
+    def save_actions(self, input: InputTriggerAction):
+        for input_action_value in input.values:
+            params = {
+                "action": self,
+                "str_cond": input_action_value.str_cond,
+                "ref_cond": input_action_value.ref_cond,
+                "bool_cond": input_action_value.bool_cond,
+            }
+            TriggerActionValue.objects.create(**params)
+
+    def get_serializer_acceptable_value(self, value=None, attr_type=None):
+        """
+        This converts TriggerActionValue to the value that EntryUpdateSerializer can accept.
+        """
+        if not attr_type:
+            attr_type = self.attr.type
+
+        value = value or self.values.first()
+        if attr_type & AttrTypeValue["array"]:
+            return [
+                self.get_serializer_acceptable_value(x, attr_type ^ AttrTypeValue["array"])
+                for x in self.values.all()
+            ]
+        elif attr_type == AttrTypeValue["boolean"]:
+            return value.bool_cond
+        elif attr_type == AttrTypeValue["named_object"]:
+            value.ref_cond.id if isinstance(value.ref_cond, Entry) else None
+            return {
+                "name": value.str_cond,
+                "id": value.ref_cond.id,
+            }
+        elif attr_type == AttrTypeValue["string"]:
+            return value.str_cond
+        elif attr_type == AttrTypeValue["text"]:
+            return value.str_cond
+        elif attr_type == AttrTypeValue["object"]:
+            return value.ref_cond.id if isinstance(value.ref_cond, Entry) else None
+
+    def run(self, user, entry, call_stacks=[]):
+        # When self.id contains in call_stacks, it means that this action is already invoked.
+        # This prevents infinite loop.
+        if self.id in call_stacks:
+            return
+
+        # update specified Entry by configured attribute value
+        setting_data = {
+            "id": entry.id,
+            "name": entry.name,
+            "attrs": [{"id": self.attr.id, "value": self.get_serializer_acceptable_value()}],
+            "delay_trigger": False,
+            "call_stacks": [*call_stacks, self.id],
+        }
+        serializer = EntryUpdateSerializer(
+            instance=entry, data=setting_data, context={"request": DRFRequest(user)}
+        )
+        if serializer:
+            serializer.is_valid(raise_exception=True)
+            serializer.save()
+
+
+class TriggerActionValue(models.Model):
+    action = models.ForeignKey(TriggerAction, on_delete=models.CASCADE, related_name="values")
+    str_cond = models.TextField(blank=True, null=True)
+    ref_cond = models.ForeignKey("entry.Entry", on_delete=models.SET_NULL, null=True, blank=True)
+    bool_cond = models.BooleanField(default=False)
+
+    # TODO: Add method to register value to Attribute when action is invoked

trigger/tests/test_api_v2.py

@@ -8,14 +8,13 @@
     TriggerCondition,
     TriggerParent,
 )
-from user.models import User
 
 
 class APITest(AironeViewTest):
     def setUp(self):
         super(APITest, self).setUp()
 
-        self.user: User = self.guest_login()
+        self.user = self.guest_login()
 
         # create Entities that are used for each tests
         self.entity_people = self.create_entity(

trigger/tests/test_models.py

@@ -20,7 +20,7 @@ class ModelTest(AironeTestCase):
     def setUp(self):
         super(ModelTest, self).setUp()
 
-        self.user: User = User.objects.create(username="test")
+        self.user = User.objects.create(username="test")
         self.entity_ref = self.create_entity(self.user, "test_entity_ref")
         self.entity = self.create_entity(
             self.user,