Merge pull request #47 from zenaton/feature/scheduling

Add scheduling

Author: MrYawe

CHANGELOG.md

@@ -6,6 +6,8 @@
 - Added a `intent_id` property when dispatching workflows and tasks, sending events to workflows, and
   pausing/resuming/killing workflows.
 
+- Added scheduling: `schedule(cron)`
+
 ### Changed
 
 ### Deprecated

zenaton/client.py

@@ -7,6 +7,7 @@
 from .abstracts.workflow import Workflow
 from .exceptions import InvalidArgumentError
 from .services.http_service import HttpService
+from .services.graphql_service import GraphQLService
 from .services.properties import Properties
 from .services.serializer import Serializer
 from .singleton import Singleton
@@ -16,6 +17,7 @@
 class Client(metaclass=Singleton):
     ZENATON_API_URL = 'https://api.zenaton.com/v1'  # Zenaton api url
     ZENATON_WORKER_URL = 'http://localhost'  # Default worker url
+    ZENATON_GATEWAY_URL = "https://gateway.zenaton.com/api"; # Zenaton gateway url
     DEFAULT_WORKER_PORT = 4001  # Default worker port
     WORKER_API_VERSION = 'v_newton'  # Default worker api version
 
@@ -49,6 +51,7 @@ def __init__(self, app_id='', api_token='', app_env=''):
         self.api_token = api_token
         self.app_env = app_env
         self.http = HttpService()
+        self.graphql = GraphQLService()
         self.serializer = Serializer()
         self.properties = Properties()
 
@@ -57,6 +60,14 @@ def __lazy_init__(self, app_id, api_token, app_env):
         self.api_token = self.api_token or api_token
         self.app_env = self.app_env or app_env
 
+    """
+        Gets the gateway url (GraphQL API)
+        :returns String the gateway url
+    """
+    def gateway_url(self):
+        url = os.environ.get('ZENATON_GATEWAY_URL') or self.ZENATON_GATEWAY_URL
+        return url
+
     """
         Gets the url for the workers
         :param String resource the endpoint for the worker
@@ -114,6 +125,41 @@ def start_task(self, task):
                     self.ATTR_MAX_PROCESSING_TIME: task.max_processing_time() if hasattr(task, 'max_processing_time') else None
                 }))
 
+    def start_scheduled_workflow(self, flow, cron):
+        url = self.gateway_url()
+        headers = self.gateway_headers()
+        query = self.graphql.CREATE_WORKFLOW_SCHEDULE
+        variables = {
+            'createWorkflowScheduleInput': {
+                'intentId': self.uuid(),
+                'environmentName': self.app_env,
+                'cron': cron,
+                'workflowName': self.class_name(flow),
+                'canonicalName': self.canonical_name(flow) or self.class_name(flow),
+                'programmingLanguage': self.PROG.upper(),
+                'properties': self.serializer.encode(self.properties.from_(flow))
+            }
+        }
+        res = self.graphql.request(url, query, variables=variables, headers=headers)
+        return res['data']['createWorkflowSchedule']
+
+    def start_scheduled_task(self, task, cron):
+        url = self.gateway_url()
+        headers = self.gateway_headers()
+        query = self.graphql.CREATE_TASK_SCHEDULE
+        variables = {
+            'createTaskScheduleInput': {
+                'intentId': self.uuid(),
+                'environmentName': self.app_env,
+                'cron': cron,
+                'taskName': self.class_name(task),
+                'programmingLanguage': self.PROG.upper(),
+                'properties': self.serializer.encode(self.properties.from_(task))
+            }
+        }
+        res = self.graphql.request(url, query, variables=variables, headers=headers)
+        return res['data']['createTaskSchedule']
+
     def update_instance(self, workflow, custom_id, mode):
         params = '{}={}'.format(self.ATTR_ID, custom_id)
         url = self.instance_worker_url(params)
@@ -212,6 +258,12 @@ def add_app_env(self, url, params):
         app_id = '{}={}&'.format(self.APP_ID, self.app_id) if self.app_id else ''
         return '{}{}{}{}'.format(url, app_env, app_id, urllib.parse.quote_plus(params, safe='=&'))
 
+    def gateway_headers(self):
+        return {'Accept': 'application/json',
+                'Content-type': 'application/json',
+                'app-id': self.app_id,
+                'api-token': self.api_token}
+
     def parse_custom_id_from(self, flow):
         custom_id = flow.id()
         if custom_id is not None:

zenaton/engine.py

@@ -29,6 +29,18 @@ def execute(self, jobs):
             return [job.handle() for job in jobs]
         return self.processor.process(jobs, True)
 
+    """
+        Executes schedule jobs synchronously
+        @param jobs [Array<Zenaton::Interfaces::Job>]
+        @param cron String
+        @return [Array<String>, nil] the results if executed locally, or nil
+    """
+    def schedule(self, jobs, cron):
+        for job in jobs:
+            Engine._check_argument(job)
+
+        [self.local_schedule(job, cron) for job in jobs]
+
     """
         Executes jobs asynchronously
         @param jobs [Array<Zenaton::Interfaces::Job>]
@@ -37,15 +49,21 @@ def execute(self, jobs):
     def dispatch(self, jobs):
         map(Engine._check_argument, jobs)
         if len(jobs) == 0 or self.processor is None:
-            [self.local_dispatch(job) for job in jobs]
+            return [self.local_dispatch(job) for job in jobs]
         if self.processor and len(jobs) > 0:
             self.processor.process(jobs, False)
 
     def local_dispatch(self, job):
         if isinstance(job, Workflow):
-            self.client.start_workflow(job)
+            return self.client.start_workflow(job)
+        else:
+            return self.client.start_task(job)
+
+    def local_schedule(self, job, cron):
+        if isinstance(job, Workflow):
+            return self.client.start_scheduled_workflow(job, cron)
         else:
-            self.client.start_task(job)
+            return self.client.start_scheduled_task(job, cron)
 
     @staticmethod
     def _check_argument(job):

zenaton/services/graphql_service.py

@@ -0,0 +1,74 @@
+from ..exceptions import ExternalError, InternalError
+import json
+import requests
+
+
+class GraphQLService:
+    CREATE_WORKFLOW_SCHEDULE = """
+        mutation ($createWorkflowScheduleInput: CreateWorkflowScheduleInput!) {
+            createWorkflowSchedule(input: $createWorkflowScheduleInput) {
+                schedule {
+                    id
+                    name
+                    cron
+                    insertedAt
+                    updatedAt
+                    target {
+                        ... on WorkflowTarget {
+                            name
+                            type
+                            canonicalName
+                            programmingLanguage
+                            properties
+                        }
+                    }
+                }
+            }
+        }
+    """
+
+    CREATE_TASK_SCHEDULE = """
+        mutation ($createTaskScheduleInput: CreateTaskScheduleInput!) {
+            createTaskSchedule(input: $createTaskScheduleInput) {
+                schedule {
+                    id
+                    name
+                    cron
+                    insertedAt
+                    updatedAt
+                    target {
+                        ... on TaskTarget {
+                            name
+                            type
+                            programmingLanguage
+                            properties
+                        }
+                    }
+                }
+            }
+        }
+    """
+
+    def request(self, url, query, variables=None, headers={}):
+        try:
+            data = {'query': query}
+            if variables:
+                data['variables'] = variables
+            r = requests.request(method='POST', url=url,
+                                 headers=headers, data=json.dumps(data))
+            if r.status_code >= 400:
+                raise InternalError(r.content)
+            content = r.json()
+            content['status_code'] = r.status_code
+
+            if 'errors' in content and isinstance(content['errors'], list) and len(content['errors']) > 0:
+                errors = content['errors']
+                for error in errors:
+                    if 'locations' in error:
+                        del error['locations']
+                raise ExternalError(errors)
+        except json.decoder.JSONDecodeError:
+            raise InternalError
+        except requests.exceptions.ConnectionError:
+            raise ConnectionError
+        return content

zenaton/traits/zenatonable.py

@@ -1,17 +1,28 @@
 from ..engine import Engine
 from ..query.builder import Builder
+from zenaton.exceptions import InvalidArgumentError
 
 
 class Zenatonable:
     """
-        Sends self as the single job to be executed
+        Sends self as the single job to be dispatched
         to the engine and returns the result
     """
     def dispatch(self):
         return Engine().dispatch([self])
 
     """
-        Sends self as the single job to be dispatched
+        Sends self as the single job to be scheduled
+        to the engine and returns the result
+    """
+    def schedule(self, cron):
+        if not isinstance(cron, str) or cron == "":
+            raise InvalidArgumentError("Param passed to 'schedule' function must be a non empty cron string")
+
+        return Engine().schedule([self], cron)
+
+    """
+        Sends self as the single job to be executed
         to the engine and returns the result
     """
     def execute(self):