str, unicode --> six.string_types

stolos/dag_tools/build.py

@@ -1,5 +1,7 @@
+from functools import reduce
 import networkx as nx
 import os
+import six
 import tempfile
 
 from stolos.exceptions import _log_raise, _log_raise_if, DAGMisconfigured
@@ -372,7 +374,7 @@ def _add_edges(dg, app_name, dep_name, dep_grp, log_details):
     except (KeyError, TypeError):
         raise DAGMisconfigured(
             "You defined a dependency but forgot to include the app_name")
-    if isinstance(parent, (unicode, str)):
+    if isinstance(parent, six.string_types):
         dg.add_edge(parent, app_name, key=dep_name, label=dep_name)
     elif isinstance(parent, cb.TasksConfigBaseSequence):
         for _parent in parent:

stolos/plugins/bash_plugin.py

@@ -1,6 +1,7 @@
 from argparse import REMAINDER
 from os import kill
 from signal import alarm, signal, SIGALRM, SIGKILL
+import six
 from subprocess import PIPE, Popen
 import sys
 
@@ -67,7 +68,7 @@ def get_bash_cmd(app_name):
             "App is not a bash job", extra=dict(
                 app_name=app_name, job_type=job_type))
     rv = meta.get('bash_cmd', '')
-    if not isinstance(rv, (str, unicode)):
+    if not isinstance(rv, six.string_types):
         log_and_raise(
             "App config for bash plugin is misconfigured:"
             " bash_cmd is not a string", dict(app_name=app_name))

stolos/plugins/pyspark_context.py

@@ -1,6 +1,7 @@
 import atexit
 import functools
 from pyspark import SparkConf, SparkContext
+import six
 import sys
 import os
 
@@ -141,12 +142,12 @@ def validate_env(app_name, env):
              " env, if supplied, must be a key: value mapping"),
             dict(app_name=app_name))
     for k, v in env.items():
-        if not isinstance(k, (str, unicode)):
+        if not isinstance(k, six.string_types):
             log_and_raise(
                 ("pyspark app misconfigured:"
                  "invalid key.  expected string"),
                 dict(app_name=app_name, key=k))
-        if not isinstance(v, (str, unicode)):
+        if not isinstance(v, six.string_types):
             log_and_raise(
                 ("pyspark app misconfigured:"
                  "invalid value.  expected string"),
@@ -158,7 +159,7 @@ def validate_uris(app_name, uris):
     msg = ("pyspark app misconfigured:"
            " %s, if supplied, must be a list of hadoop-compatible filepaths"
            ) % key
-    if not all(isinstance(x, (unicode, str)) for x in uris):
+    if not all(isinstance(x, six.string_types) for x in uris):
         log_and_raise(msg, extra=dict(app_name=app_name))
 
 
@@ -171,12 +172,13 @@ def validate_spark_conf(app_name, conf):
             dict(app_name=app_name))
 
     for k, v in conf.items():
-        if not isinstance(k, (unicode, str)):
+        if not isinstance(k, six.string_types):
             log_and_raise(
                 "pyspark app improperly configured:"
                 " Key in spark_conf must be a string",
                 dict(app_name=app_name, key=k, key_type=type(k)))
-        if not isinstance(v, (str, unicode, int, bool, float, long)):
+        if not isinstance(v, six.string_types + six.integer_types +
+                          (bool, float)):
             log_and_raise(
                 ("pyspark app improperly configured:"
                  "Value for given key in spark_conf must be an"

stolos/queue_backend/modify_job_state.py

@@ -1,4 +1,5 @@
 from os.path import join
+import six
 
 from stolos import dag_tools as dt
 from stolos import util
@@ -19,7 +20,7 @@ def _queue(app_name, job_id, queue=True, priority=None):
         extra=dict(app_name=app_name, job_id=job_id, priority=priority))
     if dt.passes_filter(app_name, job_id):
         # hack: zookeeper doesn't like unicode
-        if isinstance(job_id, unicode):
+        if isinstance(job_id, six.string_types):
             job_id = str(job_id)
         set_state(app_name, job_id, pending=True)
         if queue:

stolos/queue_backend/read_job_state.py

@@ -1,3 +1,4 @@
+import six
 from stolos import exceptions
 
 from . import shared
@@ -49,7 +50,7 @@ def check_state(app_name, job_id, raise_if_not_exists=False,
                   ignore the (pending, completed, xor failed) choice
     """
     qbcli = shared.get_qbclient()
-    if isinstance(job_id, (str, unicode)):
+    if isinstance(job_id, six.string_types):
         job_ids = [job_id]
         rvaslist = False
     else: