Merge pull request #7 from zenaton/single-task-dispatch

Single task dispatch

Author: pylebecq

CHANGELOG.md

@@ -1,25 +1,28 @@
-## Change Log
+# Changelog
 
-# Unreleased
+## Unreleased
 
-## Fixed
+### Added
+Calling `dispatch` on tasks now allows to process tasks asynchronously
+
+### Fixed
 Fixed Wait task behavior in some edge cases
 Encodes HTTP params before sending request
 
-# [0.2.5] - 2018/10/17
+## [0.2.5] - 2018/10/17
 Object Serialization (including circular structures)
 
-# [0.2.4] - 2018/09/26
+## [0.2.4] - 2018/09/26
 Enhanced WithDuration & WithTimestamp classes
 
-# [0.2.3] - 2018/09/21
+## [0.2.3] - 2018/09/21
 Minor enhancements (including the workflow find() method)
 
-# [0.2.2] - 2018/09/19
+## [0.2.2] - 2018/09/19
 New version scheme management
 
-# [0.2.1] - 2018/09/17
+## [0.2.1] - 2018/09/17
 Reorganized modules
 
-# [0.2.0] - 2018/09/14
+## [0.2.0] - 2018/09/14
 Full rewriting of the package

tests/fixtures/fixture_engine.py

@@ -4,7 +4,29 @@
 from zenaton.client import Client
 
 
+import os
+
+import pytest
+from dotenv import load_dotenv
+
+from zenaton.client import Client
+
+# LOADING CONFIG FROM .env file
+load_dotenv()
+app_id = os.getenv('ZENATON_APP_ID')
+api_token = os.getenv('ZENATON_API_TOKEN')
+app_env = os.getenv('ZENATON_APP_ENV')
+
+if not app_id:
+    raise Exception('Please include your ZENATON_APP_ID in the .env file')
+
+if not api_token:
+    raise Exception('Please include your ZENATON_API_TOKEN in the .env file')
+
+if not app_env:
+    raise Exception('Please include your ZENATON_APP_ENV in the .env file')
+
 @pytest.fixture
 def engine():
-    Client()
+    Client(app_id, api_token, app_env)
     return Engine()

tests/test_engine.py

@@ -1,4 +1,5 @@
 import pytest
+import pytest_mock
 
 from zenaton.exceptions import InvalidArgumentError
 
@@ -18,3 +19,13 @@ def test_valid_job(engine, sequential_workflow, task0):
     assert not engine.valid_job(FakeTask)
     assert engine.valid_job(sequential_workflow)
     assert engine.valid_job(task0)
+
+
+@pytest.mark.usefixtures("client", "engine", "task0")
+def test_dispatch_task(client, engine, task0, mocker):
+    mocker.spy(client, "start_task")
+    task0.dispatch()
+    assert client.start_task.call_count == 1
+
+
+

zenaton/client.py

@@ -29,6 +29,7 @@ class Client(metaclass=Singleton):
     ATTR_DATA = 'data'  # Parameter name for json payload
     ATTR_PROG = 'programming_language'  # Parameter name for the language
     ATTR_MODE = 'mode'  # Parameter name for the worker update mode
+    ATTR_MAX_PROCESSING_TIME = 'max_processing_time' # Pararameter name for the max processing time
 
     PROG = 'Python'  # The current programming language
 
@@ -87,6 +88,15 @@ def start_workflow(self, flow):
                                   self.ATTR_ID: self.parse_custom_id_from(flow)
                        }))
 
+    def start_task(self, task):
+        return self.http.post(self.worker_url('tasks'),
+                              data=json.dumps({
+                                  self.ATTR_PROG: self.PROG,
+                                  self.ATTR_NAME: self.class_name(task),
+                                  self.ATTR_DATA: self.serializer.encode(self.properties.from_(task)),
+                                  self.ATTR_MAX_PROCESSING_TIME: task.max_processing_time() if hasattr(task, 'max_processing_time') else None
+                       }))
+
     def update_instance(self, workflow, custom_id, mode):
         params = '{}={}'.format(self.ATTR_ID, custom_id)
         url = self.instance_worker_url(params)

zenaton/engine.py

@@ -44,7 +44,7 @@ def local_dispatch(self, job):
         if issubclass(job.__class__, Workflow):
             self.client.start_workflow(job)
         else:
-            job.handle()
+            self.client.start_task(job)
 
     def check_argument(self, job):
         if not self.valid_job(job):