Added timeout to rest_grouping.py

Author: AbelCheng

README.rst

@@ -421,31 +421,37 @@ Sample User Apps/Modules/Scripts
                 "[+++]": [
                     {
                         "(://)": "http://service1",
-                        "(...)": {"svc1-arg1": "arg1 of service1 payload ..." }
+                        "(...)": {"svc1-arg1": "arg1 of service1 payload ..." },
+                        "(:/!)": 600
                     },
                     {
                         "(://)": "http://service2",
-                        "(.|.)": {"svc2-arg1": "arg1 of service2 payload ..." }
+                        "(.|.)": {"svc2-arg1": "arg1 of service2 payload ..." },
+                        "(:/!)": 600
                     },
                     {
                         "[###]": [
                             {
                                 "(://)": "http://service3",
-                                "(...)": {"svc3-arg1": "arg1 of service3 payload ..." }
+                                "(...)": {"svc3-arg1": "arg1 of service3 payload ..." },
+                                "(:/!)": 1800
                             },
                             {
                                 "(://)": "http://service4",
-                                "(...)": {"svc4-arg1": "arg1 of service4 payload ..." }
+                                "(...)": {"svc4-arg1": "arg1 of service4 payload ..." },
+                                "(:/!)": 1800
                             },
                             {
                                 "(://)": "http://service5",
-                                "(...)": {"svc5-arg1": "arg1 of service5 payload ..." }
+                                "(...)": {"svc5-arg1": "arg1 of service5 payload ..." },
+                                "(:/!)": 1800
                             }
                         ]
                     },
                     {
                         "(://)": "http://service6",
-                        "(...)": {"svc6-arg1": "arg1 of service6 payload ..." }
+                        "(...)": {"svc6-arg1": "arg1 of service6 payload ..." },
+                        "(:/!)": 600
                     }
                 ]
             }
@@ -459,13 +465,14 @@ Sample User Apps/Modules/Scripts
 
         #.  Single Service (Leaf Service)
 
-            This is the most basic unit that constitutes a service group (virtual service). It requires a URL and a dictionary of arguments as the payload:
+            This is the most basic unit that constitutes a service group (virtual service). It requires a URL, a dictionary of arguments as the payload, and an optional timeout seconds:
 
             .. code-block:: JSON
 
                 {
                     "(://)": "http://service1",
-                    "(...)": {"svc1-arg1": "arg1 of service1 payload ..." }
+                    "(...)": {"svc1-arg1": "arg1 of service1 payload ..." },
+                    "(:/!)": timeout seconds
                 }
 
             Or
@@ -474,12 +481,14 @@ Sample User Apps/Modules/Scripts
 
                 {
                     "(://)": "http://service2",
-                    "(.|.)": {"svc2-arg1": "arg1 of service2 payload ..." }
+                    "(.|.)": {"svc2-arg1": "arg1 of service2 payload ..." },
+                    "(:/!)": timeout seconds
                 }
 
-            +   "``(://)``" - *Key : Value* - URL of the service call
-            +   "``(...)``" - *Key : Value* - a dictionary of arguments (payload) for the service call
-            +   "``(.|.)``" - *Key : Value* - merge the results of the previous service as pipeline arguments into the current arguments
+            +   "``(://)``" - *Key : Value* - "URL of the service call"
+            +   "``(...)``" - *Key : Value* - {A dictionary of arguments (payload) for the service call}
+            +   "``(.|.)``" - *Key : Value* - {Merge the results of the previous service as pipeline arguments into this dictionary of arguments}
+            +   "``(:/!)``" - *Key : Value* - timeout seconds (optional) How many seconds to wait for the REST service to respond before giving up
 
             Each service is an executable/callable unit, let's have a convention to use a rounded rectangle as its graphical symbol.
 

Sample/UserApps/ServicesGrouping/rest_grouping.py

@@ -17,11 +17,16 @@
 from simple_rest_call import request_json
 
 
-_reserved_key_parallel_group : str = '[###]'
-_reserved_key_serial_group : str = '[+++]'
-_reserved_key_rest_url : str = '(://)'
-_reserved_key_payload : str = '(...)'
-_reserved_key_payload_with_pipe : str = '(.|.)'
+_reserved_key_parallel_group : str = "[###]"
+_reserved_key_serial_group : str = "[+++]"
+_reserved_key_rest_url : str = "(://)"
+_reserved_key_payload : str = "(...)"
+_reserved_key_payload_with_pipe : str = "(.|.)"
+_reserved_key_timeout : str = "(:/!)"
+
+
+def _task_func(url:str, data:dict=None, timeout:float=None):
+    return request_json(url, data, timeout=timeout)
 
 
 def _pipeargs_merge_fn(kw_args:Dict[str, Any], pipe_args:Dict[str, Any]) -> Dict[str, Any]:
@@ -43,13 +48,10 @@ class RestTaskLoader(ITaskLoader):
     """This class is used to load a group of RESTful services (call tasks) from a JSON payload into ``TaskContainer``"""
     def __init__(self, thread_pool:ThreadPoolExecutor):
         self.thread_pool = thread_pool
-        self.func = request_json
-        self.merge_fn = _pipeargs_merge_fn
-        self.timeout = None
 
 
     def create_base_container(self) -> TaskContainer:
-        return TaskContainer(self.func, self.merge_fn, self.thread_pool, **{'timeout':self.timeout})
+        return TaskContainer(_task_func, _pipeargs_merge_fn, self.thread_pool)
 
 
     def extract_single_task(self, task_node:Dict[str, Any]) -> Tuple[tuple, Dict[str, Any], bool]:
@@ -67,7 +69,9 @@ def extract_single_task(self, task_node:Dict[str, Any]) -> Tuple[tuple, Dict[str
             else:
                 with_pipe = False
 
-            return ((), {'url': url, 'data': data}, with_pipe)
+            timeout = task_node.get(_reserved_key_timeout)
+
+            return ((), {'url': url, 'data': data, 'timeout': timeout}, with_pipe)
         else:
             return None
 
@@ -80,10 +84,15 @@ def extract_parallel_group(self, task_node:Dict[str, Any]) -> List[Dict[str, Any
         return task_node.get(_reserved_key_parallel_group)
 
 
+    def load(self, task_tree:Dict[str, Any]) -> TaskContainer:
+        container = super().load(task_tree)
+        container.timeout = task_tree.get(_reserved_key_timeout)
+        return container
+
 
 def start(rest:Dict[str, Any]):
     """the main entry for the client to call a group of RESTful services."""
-    with ThreadPoolExecutor(max_workers=32) as thread_pool:
+    with ThreadPoolExecutor(max_workers=64) as thread_pool:
         loader = RestTaskLoader(thread_pool)
         container = loader.load(rest)
         return container.run()

Utilities.PyPI/task_grouping.py

@@ -14,6 +14,10 @@
 
     This is an abstract base class for implementing a concrete loader class to load a task tree from Dict/JSON to ``TaskContainer``.
 
+For detailed usage, see the practice sample code: https://github.com/DataBooster/PyWebApi/blob/master/Sample/UserApps/ServicesGrouping/rest_grouping.py
+
+and its product documentation: https://github.com/DataBooster/PyWebApi#services-grouping
+
 ----
 
 | Homepage and documentation: https://github.com/DataBooster/PyWebApi
@@ -290,4 +294,4 @@ def load(self, task_tree:Dict[str, Any]) -> TaskContainer:
 
 
 
-__version__ = "0.1a1"
+__version__ = "0.1a2"