DPNLPF-1300: make framework async

core/basic_models/actions/basic_actions.py

@@ -1,4 +1,5 @@
 # coding: utf-8
+import asyncio
 import random
 from typing import Union, Dict, List, Any, Optional
 
@@ -25,8 +26,8 @@ def __init__(self, items: Dict[str, Any], id: Optional[str] = None):
         self.id = id
         self.version = items.get("version", -1)
 
-    def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
-            params: Optional[Dict[str, Union[str, float, int]]] = None) -> Optional[List[Command]]:
+    async def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
+                  params: Optional[Dict[str, Union[str, float, int]]] = None) -> Optional[List[Command]]:
         raise NotImplementedError
 
     def on_run_error(self, text_preprocessing_result, user):
@@ -51,9 +52,9 @@ def __init__(self, items: Dict[str, Any], id: Optional[str] = None):
         self.request_type = items.get("request_type") or self.DEFAULT_REQUEST_TYPE
         self.request_data = items.get("request_data")
 
-    def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
-            params: Optional[Dict[str, Union[str, float, int]]] = None) -> Optional[List[Command]]:
-        super(CommandAction, self).run(user, text_preprocessing_result, params)
+    async def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
+                  params: Optional[Dict[str, Union[str, float, int]]] = None) -> Optional[List[Command]]:
+        await super(CommandAction, self).run(user, text_preprocessing_result, params)
         return None
 
 
@@ -66,8 +67,8 @@ def __init__(self, items: Dict[str, Any], id: Optional[str] = None):
         super(DoingNothingAction, self).__init__(items, id)
         self.nodes = items.get("nodes") or {}
 
-    def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
-            params: Optional[Dict[str, Union[str, float, int]]] = None) -> List[Command]:
+    async def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
+                  params: Optional[Dict[str, Union[str, float, int]]] = None) -> List[Command]:
         commands = [Command(self.command, self.nodes, self.id, request_type=self.request_type,
                             request_data=self.request_data)]
         return commands
@@ -98,11 +99,56 @@ def build_requirement(self):
     def build_internal_item(self):
         return self._item
 
-    def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
-            params: Optional[Dict[str, Union[str, float, int]]] = None) -> Optional[List[Command]]:
+    async def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
+                  params: Optional[Dict[str, Union[str, float, int]]] = None) -> Optional[List[Command]]:
         result = None
-        if self.requirement.check(text_preprocessing_result, user, params):
-            result = self.internal_item.run(user, text_preprocessing_result, params)
+        if await self.requirement.check(text_preprocessing_result, user, params):
+            result = await self.internal_item.run(user, text_preprocessing_result, params)
+        return result
+
+
+class GatherChoiceAction(Action):
+    version: Optional[int]
+    requirement_actions: RequirementAction
+    else_action: Action
+
+    FIELD_REQUIREMENT_KEY = "requirement_actions"
+    FIELD_ELSE_KEY = "else_action"
+
+    def __init__(self, items: Dict[str, Any], id: Optional[str] = None):
+        super(GatherChoiceAction, self).__init__(items, id)
+        self._requirement_items = items[self.FIELD_REQUIREMENT_KEY]
+        self._else_item = items.get(self.FIELD_ELSE_KEY)
+
+        self.items = self.build_items()
+
+        if self._else_item:
+            self.else_item = self.build_else_item()
+        else:
+            self.else_item = None
+
+    @list_factory(RequirementAction)
+    def build_items(self):
+        return self._requirement_items
+
+    @factory(Action)
+    def build_else_item(self):
+        return self._else_item
+
+    async def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
+                  params: Optional[Dict[str, Union[str, float, int]]] = None) -> Optional[List[Command]]:
+        result = None
+        choice_is_made = False
+        check_results = await asyncio.gather(
+            item.requirement.check(text_preprocessing_result, user, params) for item in self.items)
+        for i, checked in enumerate(check_results):
+            if checked:
+                item = self.items[i]
+                result = await item.internal_item.run(user, text_preprocessing_result, params)
+                choice_is_made = True
+                break
+        if not choice_is_made and self._else_item:
+            result = await self.else_item.run(user, text_preprocessing_result, params)
         return result
 
 
@@ -134,18 +180,17 @@ def build_items(self):
     def build_else_item(self):
         return self._else_item
 
-    def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
-            params: Optional[Dict[str, Union[str, float, int]]] = None) -> Optional[List[Command]]:
+    async def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
+                  params: Optional[Dict[str, Union[str, float, int]]] = None) -> Optional[List[Command]]:
         result = None
         choice_is_made = False
         for item in self.items:
-            checked = item.requirement.check(text_preprocessing_result, user, params)
-            if checked:
-                result = item.internal_item.run(user, text_preprocessing_result, params)
+            if await item.requirement.check(text_preprocessing_result, user, params):
+                result = await item.internal_item.run(user, text_preprocessing_result, params)
                 choice_is_made = True
                 break
         if not choice_is_made and self._else_item:
-            result = self.else_item.run(user, text_preprocessing_result, params)
+            result = await self.else_item.run(user, text_preprocessing_result, params)
         return result
 
 
@@ -182,13 +227,13 @@ def build_item(self):
     def build_else_item(self):
         return self._else_item
 
-    def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
-            params: Optional[Optional[Dict[str, Union[str, float, int]]]] = None) -> Optional[List[Command]]:
+    async def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
+                  params: Optional[Optional[Dict[str, Union[str, float, int]]]] = None) -> Optional[List[Command]]:
         result = None
-        if self.requirement.check(text_preprocessing_result, user, params):
-            result = self.item.run(user, text_preprocessing_result, params)
+        if await self.requirement.check(text_preprocessing_result, user, params):
+            result = await self.item.run(user, text_preprocessing_result, params)
         elif self._else_item:
-            result = self.else_item.run(user, text_preprocessing_result, params)
+            result = await self.else_item.run(user, text_preprocessing_result, params)
         return result
 
 
@@ -205,11 +250,11 @@ def __init__(self, items: Dict[str, Any], id: Optional[str] = None):
     def build_actions(self):
         return self._actions
 
-    def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
-            params: Optional[Dict[str, Union[str, float, int]]] = None) -> Optional[List[Command]]:
+    async def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
+                  params: Optional[Dict[str, Union[str, float, int]]] = None) -> Optional[List[Command]]:
         commands = []
         for action in self.actions:
-            action_result = action.run(user, text_preprocessing_result, params)
+            action_result = await action.run(user, text_preprocessing_result, params)
             if action_result:
                 commands += action_result
         return commands
@@ -225,8 +270,8 @@ def __init__(self, items: Dict[str, Any], id: Optional[str] = None):
         self._actions_count = len(items["actions"])
         self._last_action_ids_storage = items["last_action_ids_storage"]
 
-    def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
-            params: Optional[Dict[str, Union[str, float, int]]] = None) -> Optional[List[Command]]:
+    async def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
+                  params: Optional[Dict[str, Union[str, float, int]]] = None) -> Optional[List[Command]]:
         last_ids = user.last_action_ids[self._last_action_ids_storage]
         all_indexes = list(range(self._actions_count))
         max_last_ids_count = self._actions_count - 1
@@ -236,7 +281,7 @@ def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingRe
         action_index = random.choice(available_indexes)
         action = self.actions[action_index]
         last_ids.add(action_index)
-        result = action.run(user, text_preprocessing_result, params)
+        result = await action.run(user, text_preprocessing_result, params)
         return result
 
 
@@ -251,8 +296,8 @@ def __init__(self, items, id=None):
     def build_actions(self):
         return self._raw_actions
 
-    def run(self, user, text_preprocessing_result, params=None):
+    async def run(self, user, text_preprocessing_result, params=None):
         pos = random.randint(0, len(self._raw_actions) - 1)
         action = self.actions[pos]
-        command_list = action.run(user, text_preprocessing_result, params=params)
+        command_list = await action.run(user, text_preprocessing_result, params=params)
         return command_list

core/basic_models/actions/client_profile_actions.py

@@ -54,8 +54,8 @@ def __init__(self, items: Dict[str, Any], id: Optional[str] = None):
             {"memoryPartition": key, "tags": val} for key, val in self._nodes["memory"].items()
         ]
 
-    def run(self, user: User, text_preprocessing_result: BaseTextPreprocessingResult,
-            params: Optional[Dict[str, Union[str, float, int]]] = None) -> Optional[List[Command]]:
+    async def run(self, user: User, text_preprocessing_result: BaseTextPreprocessingResult,
+                  params: Optional[Dict[str, Union[str, float, int]]] = None) -> Optional[List[Command]]:
         self._nodes["consumer"] = {"projectId": user.settings["template_settings"]["project_id"]}
 
         settings_kafka_key = user.settings["template_settings"].get("client_profile_kafka_key")
@@ -70,7 +70,7 @@ def run(self, user: User, text_preprocessing_result: BaseTextPreprocessingResult
             user.behaviors.add(user.message.generate_new_callback_id(), self.behavior, scenario_id,
                                text_preprocessing_result.raw, action_params=pickle_deepcopy(params))
 
-        commands = super().run(user, text_preprocessing_result, params)
+        commands = await super().run(user, text_preprocessing_result, params)
         return commands
 
 
@@ -144,8 +144,8 @@ def __init__(self, items: Dict[str, Any], id: Optional[str] = None):
         self.kafka_key = items.get("kafka_key")
         self._nodes["root_nodes"] = {"protocolVersion": items.get("protocolVersion") or 3}
 
-    def run(self, user: User, text_preprocessing_result: BaseTextPreprocessingResult,
-            params: Optional[Dict[str, Union[str, float, int]]] = None) -> Optional[List[Command]]:
+    async def run(self, user: User, text_preprocessing_result: BaseTextPreprocessingResult,
+                  params: Optional[Dict[str, Union[str, float, int]]] = None) -> Optional[List[Command]]:
         self._nodes["consumer"] = {"projectId": user.settings["template_settings"]["project_id"]}
 
         settings_kafka_key = user.settings["template_settings"].get("client_profile_kafka_key")
@@ -155,5 +155,5 @@ def run(self, user: User, text_preprocessing_result: BaseTextPreprocessingResult
             "kafka_replyTopic": user.settings["template_settings"]["consumer_topic"]
         }
 
-        commands = super().run(user, text_preprocessing_result, params)
+        commands = await super().run(user, text_preprocessing_result, params)
         return commands

core/basic_models/actions/counter_actions.py

@@ -18,20 +18,20 @@ def __init__(self, items: Dict[str, Any], id: Optional[str] = None):
         self.value = items.get("value", 1)
         self.lifetime = items.get("lifetime")
 
-    def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
-            params: Optional[Dict[str, Union[str, float, int]]] = None) -> None:
+    async def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
+                  params: Optional[Dict[str, Union[str, float, int]]] = None) -> None:
         user.counters[self.key].inc(self.value, self.lifetime)
 
 
 class CounterDecrementAction(CounterIncrementAction):
-    def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
-            params: Optional[Dict[str, Union[str, float, int]]] = None) -> None:
+    async def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
+                  params: Optional[Dict[str, Union[str, float, int]]] = None) -> None:
         user.counters[self.key].dec(-self.value, self.lifetime)
 
 
 class CounterClearAction(CounterIncrementAction):
-    def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
-            params: Optional[Dict[str, Union[str, float, int]]] = None) -> None:
+    async def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
+                  params: Optional[Dict[str, Union[str, float, int]]] = None) -> None:
         user.counters.clear(self.key)
 
 
@@ -48,8 +48,8 @@ def __init__(self, items: Dict[str, Any], id: Optional[str] = None):
         self.reset_time = items.get("reset_time", False)
         self.time_shift = items.get("time_shift", 0)
 
-    def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
-            params: Optional[Dict[str, Union[str, float, int]]] = None) -> None:
+    async def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
+                  params: Optional[Dict[str, Union[str, float, int]]] = None) -> None:
         user.counters[self.key].set(self.value, self.reset_time, self.time_shift)
 
 
@@ -61,7 +61,7 @@ def __init__(self, items: Dict[str, Any], id: Optional[str] = None):
         self.reset_time = items.get("reset_time", False)
         self.time_shift = items.get("time_shift", 0)
 
-    def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
-            params: Optional[Dict[str, Union[str, float, int]]] = None) -> None:
+    async def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
+                  params: Optional[Dict[str, Union[str, float, int]]] = None) -> None:
         value = user.counters[self.src].value
         user.counters[self.dst].set(value, self.reset_time, self.time_shift)

core/basic_models/actions/external_actions.py

@@ -24,8 +24,8 @@ def __init__(self, items: Dict[str, Any], id: Optional[str] = None):
         super(ExternalAction, self).__init__(items, id)
         self._action_key = items["action"]
 
-    def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
-            params: Optional[Dict[str, Union[str, float, int]]] = None) -> List[Command]:
+    async def run(self, user: BaseUser, text_preprocessing_result: BaseTextPreprocessingResult,
+                  params: Optional[Dict[str, Union[str, float, int]]] = None) -> List[Command]:
         action = user.descriptions["external_actions"][self._action_key]
-        commands = action.run(user, text_preprocessing_result, params)
+        commands = await action.run(user, text_preprocessing_result, params)
         return commands

core/basic_models/actions/push_action.py

@@ -64,8 +64,8 @@ def _render_request_data(self, action_params):
         }
         return request_data
 
-    def run(self, user: User, text_preprocessing_result: BaseTextPreprocessingResult,
-            params: Optional[Dict[str, Union[str, float, int]]] = None) -> List[Command]:
+    async def run(self, user: User, text_preprocessing_result: BaseTextPreprocessingResult,
+                  params: Optional[Dict[str, Union[str, float, int]]] = None) -> List[Command]:
         params = params or {}
         command_params = {
             "surface": self.surface,