Add basic XML tag generation

MANIFEST.in

@@ -0,0 +1 @@
+include README.md LICENSE

Makefile

@@ -24,5 +24,6 @@ autopep8:
 lint:
 		@echo 'Linting...'
 		@pylint --rcfile=pylintrc pyoozie tests
+		@flake8
 
 autolint: autopep8 lint

dev_requirements.txt

@@ -1,5 +1,8 @@
 pytest>=2.7
 pytest-cov
 pytest-randomly
+pytest-mock
 pylint
+xmltodict
 autopep8
+flake8

pylintrc

@@ -35,8 +35,19 @@ load-plugins=
 # it should appear only once).
 
 # CHANGED:
+# too-few-public-methods
+# too-many-instance-attributes
+# too-many-arguments
+# protected-access; _xml() calls
+# invalid-name; some 2-letter variables are OK
+# redefined-outer-name; doesn't play well with pytest fixtures
+# fixme; we're in-development, so we need some of these
 # C0111: Missing docstring
-disable=C0111
+# D101 Missing docstring in public class
+# D102 Missing docstring in public method
+# D105 Missing docstring in magic method
+disable=C0111,D101,D102,D105,too-few-public-methods,too-many-instance-attributes,too-many-arguments,
+ protected-access,invalid-name,redefined-outer-name,fixme
 
 
 [REPORTS]

pyoozie/__init__.py

@@ -1,4 +1,21 @@
 # Copyright (c) 2017 "Shopify inc." All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be found in the LICENSE file.
 
+from pyoozie.coordinator import Coordinator, ExecutionOrder
+from pyoozie.tags import Parameters, Configuration, Credentials, Shell, SubWorkflow, GlobalConfiguration, \
+    Email, IdentifierTooLongError
+from pyoozie.builder import WorkflowBuilder, CoordinatorBuilder
+
 __version__ = '0.0.0'
+
+__all__ = (
+    # coordinator
+    'Coordinator', 'ExecutionOrder', 'Configuration', 'Parameters',
+
+    # tags
+    'Parameters', 'Configuration', 'Credentials', 'Shell', 'SubWorkflow', 'GlobalConfiguration', \
+    'Email', 'IdentifierTooLongError',
+
+    # builder
+    'WorkflowBuilder', 'CoordinatorBuilder',
+)

pyoozie/builder.py

@@ -0,0 +1,104 @@
+# Copyright (c) 2017 "Shopify inc." All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be found in the LICENSE file.
+from __future__ import unicode_literals
+
+from pyoozie.coordinator import Coordinator
+from pyoozie.tags import Configuration
+
+
+def _workflow_submission_xml(username, workflow_xml_path, configuration=None, indent=False):
+    """Generate a Workflow XML submission message to POST to Oozie."""
+    submission = Configuration(configuration)
+    submission.update({
+        'user.name': username,
+        'oozie.wf.application.path': workflow_xml_path,
+    })
+    return submission.xml(indent)
+
+
+def _coordinator_submission_xml(username, coord_xml_path, configuration=None, indent=False):
+    """Generate a Coordinator XML submission message to POST to Oozie."""
+    submission = Configuration(configuration)
+    submission.update({
+        'user.name': username,
+        'oozie.coord.application.path': coord_xml_path,
+    })
+    return submission.xml(indent)
+
+
+class WorkflowBuilder(object):
+
+    def __init__(self, name):
+        # Initially, let's just use a static template and only one action payload and one action on error
+        self._name = name
+        self._action_name = None
+        self._action_payload = None
+        self._action_error = None
+        self._kill_message = None
+
+    def add_action(self, name, action, action_on_error, kill_on_error='${wf:lastErrorNode()} - ${wf:id()}'):
+        # Today you can't rename your action and you can only have one, but in the future you can add multiple
+        # named actions
+        if any((self._action_name, self._action_payload, self._action_error, self._kill_message)):
+            raise NotImplementedError("Can only add one action in this version")
+        else:
+            self._action_name = name
+            self._action_payload = action
+            self._action_error = action_on_error
+            self._kill_message = kill_on_error
+
+        return self
+
+    def build(self, indent=False):
+        def format_xml(xml):
+            xml = xml.replace("<?xml version='1.0' encoding='UTF-8'?>", '')
+            return '\n'.join([(' ' * 8) + line for line in xml.strip().split('\n')])
+        return '''
+<?xml version="1.0" encoding="UTF-8"?>
+<workflow-app xmlns="uri:oozie:workflow:0.5"
+              name="{name}">
+    <start to="action-{action_name}" />
+    <action name="action-{action_name}">
+{action_payload_xml}
+        <ok to="end" />
+        <error to="action-error" />
+    </action>
+    <action name="action-error">
+{action_error_xml}
+        <ok to="kill" />
+        <error to="kill" />
+    </action>
+    <kill name="kill">
+        <message>{kill_message}</message>
+    </kill>
+    <end name="end" />
+</workflow-app>
+'''.format(action_payload_xml=format_xml(self._action_payload.xml(indent=indent)),
+           action_error_xml=format_xml(self._action_error.xml(indent=indent)),
+           kill_message=self._kill_message,
+           action_name=self._action_name,
+           name=self._name).strip()
+
+
+class CoordinatorBuilder(object):
+
+    def __init__(self, name, workflow_xml_path, frequency_in_minutes, start, end=None, timezone=None,
+                 workflow_configuration=None, timeout_in_minutes=None, concurrency=None, execution_order=None,
+                 throttle=None, parameters=None):
+        self._coordinator = Coordinator(
+            name=name,
+            workflow_app_path=workflow_xml_path,
+            frequency=frequency_in_minutes,
+            start=start,
+            end=end,
+            timezone=timezone,
+            workflow_configuration=workflow_configuration,
+            timeout=timeout_in_minutes,
+            concurrency=concurrency,
+            execution_order=execution_order,
+            throttle=throttle,
+            parameters=parameters,
+        )
+
+    def build(self, indent=False):
+        return self._coordinator.xml(indent)

pyoozie/coordinator.py

@@ -0,0 +1,88 @@
+# Copyright (c) 2017 "Shopify inc." All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be found in the LICENSE file.
+from __future__ import unicode_literals
+
+from datetime import timedelta
+from enum import Enum
+
+from pyoozie.tags import _validate, Xml, Parameters, Configuration
+
+
+class ExecutionOrder(Enum):
+    """Execution order used for coordinator jobs."""
+
+    FIFO = 'FIFO'
+    LIFO = 'LIFO'
+    LAST_ONLY = 'LAST_ONLY'
+    NONE = 'NONE'
+
+    def __str__(self):
+        return self.value
+
+
+def format_datetime(value):
+    return value.strftime('%Y-%m-%dT%H:%MZ')
+
+
+class Coordinator(Xml):
+
+    def __init__(self, name, workflow_app_path, frequency, start, end=None, timezone=None,
+                 workflow_configuration=None, timeout=None, concurrency=None, execution_order=None, throttle=None,
+                 parameters=None):
+        super(Coordinator, self).__init__('coordinator-app')
+        # Compose and validate dates/frequencies
+        if end is None:
+            end = start + timedelta(days=100 * 365.24)
+        assert end > start, "End time (%s) must be greater than the start time (%s)" % \
+            (format_datetime(end), format_datetime(start))
+        assert frequency >= 5, "Frequency (%d min) must be greater than or equal to 5 min" % frequency
+
+        # Coordinator
+        self.name = _validate(name)
+        self.frequency = frequency
+        self.start = start
+        self.end = end
+        self.timezone = timezone if timezone else 'UTC'
+
+        # Workflow action
+        self.workflow_app_path = workflow_app_path
+        self.workflow_configuration = Configuration(workflow_configuration)
+
+        # Controls
+        self.timeout = timeout
+        self.concurrency = concurrency
+        self.execution_order = execution_order
+        self.throttle = throttle
+
+        self.parameters = Parameters(parameters)
+
+    def _xml(self, doc, tag, text):
+        with tag(self.xml_tag, xmlns="uri:oozie:coordinator:0.5", name=self.name, frequency=str(self.frequency),
+                 start=format_datetime(self.start), end=format_datetime(self.end), timezone=self.timezone):
+
+            if self.parameters:
+                self.parameters._xml(doc, tag, text)
+
+            if self.timeout or self.concurrency or self.execution_order or self.throttle:
+                with tag('controls'):
+                    if self.timeout:
+                        with tag('timeout'):
+                            text(str(self.timeout))
+                    if self.concurrency:
+                        with tag('concurrency'):
+                            text(str(self.concurrency))
+                    if self.execution_order:
+                        with tag('execution'):
+                            text(str(self.execution_order))
+                    if self.throttle:
+                        with tag('throttle'):
+                            text(str(self.throttle))
+
+            with tag('action'):
+                with tag('workflow'):
+                    with tag('app-path'):
+                        text(self.workflow_app_path)
+                    if self.workflow_configuration:
+                        self.workflow_configuration._xml(doc, tag, text)
+
+        return doc