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,18 @@ 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 validate, Parameters, Configuration, Credentials, Shell, SubWorkflow, GlobalConfiguration, \
+    Email, IdentifierTooLongError
+from pyoozie.builder import workflow, coordinator, _coordinator_submission_xml, _workflow_submission_xml
+
 __version__ = '0.0.0'
+
+__all__ = (
+    # coordinator
+    'Coordinator', 'ExecutionOrder', 'Configuration', 'Parameters',
+
+    # tags
+    'validate', 'Parameters', 'Configuration', 'Credentials', 'Shell', 'SubWorkflow', 'GlobalConfiguration', \
+    'Email', 'IdentifierTooLongError',
+
+    # builder
+    'workflow', 'coordinator', '_coordinator_submission_xml', '_workflow_submission_xml',
+)

pyoozie/builder.py

@@ -0,0 +1,135 @@
+# 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, Parameters
+
+
+def _workflow_submission_xml(hadoop_user, hdfs_path, configuration=None, indent=False):
+    """Generate a Workflow XML submission message to POST to Oozie."""
+    submission = Configuration(configuration)
+    submission.update({
+        'user.name': hadoop_user,
+        'oozie.wf.application.path': hdfs_path
+    })
+    return submission.xml(indent)
+
+
+def _coordinator_submission_xml(hadoop_user, hdfs_path, configuration=None, indent=False):
+    """Generate a Coordinator XML submission message to POST to Oozie."""
+    submission = Configuration(configuration)
+    submission.update({
+        'user.name': hadoop_user,
+        'oozie.coord.application.path': hdfs_path,
+    })
+    return submission.xml(indent)
+
+
+class workflow(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 self._action_name is None and self._action_payload is None and self._action_error is None and \
+           self._kill_message is None:
+            self._action_name = name
+            self._action_payload = action
+            self._action_error = action_on_error
+            self._kill_message = kill_on_error
+        else:
+            raise NotImplementedError("Can only add one action in this version")
+        return self
+
+    def build(self, indent=False):
+        def remove_header(xml):
+            return xml.replace("<?xml version='1.0' encoding='UTF-8'?>", '')
+        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=remove_header(self._action_payload.xml(indent=indent)),
+           action_error_xml=remove_header(self._action_error.xml(indent=indent)),
+           kill_message=self._kill_message,
+           action_name=self._action_name,
+           name=self._name).strip()
+
+    def submit(self, oozie_url, hdfs_path, hadoop_user, hdfs_callback, timeout_in_seconds=None,
+               verbose=False, start=False, indent=False):
+        xml = self.build(indent=indent)
+        hdfs_callback(hdfs_path, xml)
+        # TODO create Oozie API and submit
+        from mock import Mock
+        OozieAPI = Mock()
+        api = OozieAPI(url=oozie_url, user=hadoop_user, timeout=timeout_in_seconds, verbose=verbose)
+        api.jobs_submit_workflow(hdfs_path=hdfs_path, start=start)
+        raise NotImplementedError()
+
+
+class coordinator(object):
+
+    def __init__(self, name, workflow, frequency_in_minutes, start, end=None, timezone=None,
+                 workflow_configuration=None, timeout_in_minutes=None, concurrency=None, execution_order=None,
+                 throttle=None, parameters=None):
+        workflow_configuration = Configuration(workflow_configuration) if workflow_configuration else None
+        self._coordinator = Coordinator(
+            name=name,
+            workflow_app_path=None,  # Defer this until the build
+            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(parameters) if parameters else None
+        )
+        self._workflow = workflow
+
+    def build(self, workflow_hdfs_path, indent=False):
+        self._coordinator.workflow_app_path = workflow_hdfs_path
+        return self._coordinator.xml(indent)
+
+    def submit(self, oozie_url, workflow_hdfs_path, coord_hdfs_path, hadoop_user, hdfs_callback,
+               timeout_in_seconds=None, verbose=False, indent=False):
+        self._workflow.submit(
+            oozie_url=oozie_url,
+            hdfs_path=workflow_hdfs_path,
+            hadoop_user=hadoop_user,
+            hdfs_callback=hdfs_callback,
+            start=False,
+            indent=False)
+        xml = self.build(workflow_hdfs_path, indent=indent)
+        hdfs_callback(coord_hdfs_path, xml)
+        # TODO create Oozie API and submit
+        from mock import Mock
+        OozieAPI = Mock()
+        api = OozieAPI(url=oozie_url, user=hadoop_user, timeout=timeout_in_seconds, verbose=verbose)
+        api.job_submit_coordinator(hdfs_path=coord_hdfs_path)
+        raise NotImplementedError()

pyoozie/coordinator.py

@@ -0,0 +1,102 @@
+# 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 Xml, Parameters, validate
+
+
+class ExecutionOrder(Enum):
+    """Execution order used for coordinator jobs."""
+
+    FIFO = 'FIFO'
+
+    LIFO = 'LIFO'
+
+    LAST_ONLY = 'LAST_ONLY'
+    # "When LAST_ONLY is set, an action that is WAITING or READY will be SKIPPED when the current time is past the
+    # next action's nominal time.  For example, suppose action 1 and 2 are both WAITING , the current time is
+    # 5:00pm, and action 2's nominal time is 5:10pm. In 10 minutes from now, at 5:10pm, action 1 will become
+    # SKIPPED, assuming it doesn't transition to SUBMITTED (or a terminal state) before then. Another way of
+    # thinking about this is to view it as similar to setting the timeout equal to the frequency, except that the
+    # SKIPPED status doesn't cause the coordinator job to eventually become DONEWITHERROR and can actually become
+    # SUCCEEDED (i.e. it's a "good" version of TIMEDOUT ). LAST_ONLY is useful if you want a recurring job, but do
+    # not actually care about the individual instances and just always want the latest action."
+
+    NONE = 'NONE'
+    # "Similar to LAST_ONLY except all older materializations are skipped. When NONE is set, an action that is
+    # WAITING or READY will be SKIPPED when the current time is more than a certain configured number of minutes
+    # (tolerance) past the action's nominal time."
+
+    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 = 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