Experimental python3 support

Author: tume

examples/simple_crawler/crawler.py

@@ -86,16 +86,16 @@ def run(self, params):
 
         collection = connections.mongodb_jobs.simple_crawler_urls
 
-        print
-        print "Crawl stats"
-        print "==========="
-        print "URLs queued: %s" % collection.find().count()
-        print "URLs successfully crawled: %s" % collection.find({"fetched_date": {"$exists": True}}).count()
-        print "URLs redirected: %s" % collection.find({"redirected_to": {"$exists": True}}).count()
-        print "Bytes fetched: %s" % (list(collection.aggregate(
+        print()
+        print( "Crawl stats")
+        print( "===========")
+        print( "URLs queued: %s" % collection.find().count())
+        print( "URLs successfully crawled: %s" % collection.find({"fetched_date": {"$exists": True}}).count())
+        print( "URLs redirected: %s" % collection.find({"redirected_to": {"$exists": True}}).count())
+        print( "Bytes fetched: %s" % (list(collection.aggregate(
             {"$group": {"_id": None, "sum": {"$sum": "$html_length"}}}
-        )) or [{}])[0].get("sum", 0)
-        print
+        )) or [{}])[0].get("sum", 0))
+        print()
 
 
 class Reset(Task):

mrq/basetasks/utils.py

@@ -1,3 +1,4 @@
+from __future__ import print_function
 from mrq.task import Task
 from mrq.queue import Queue
 from bson import ObjectId
@@ -50,7 +51,7 @@ def build_query(self):
         if self.params.get("params"):
             params_dict = json.loads(self.params.get("params"))  # pylint: disable=no-member
 
-            for key in params_dict.keys():
+            for key in list(params_dict.keys()):
                 query["params.%s" % key] = params_dict[key]
 
         return query
@@ -76,7 +77,7 @@ def perform_action(self, action, query, destination_queue):
             else:
 
                 tasks_defs = get_current_config().get("tasks", {})
-                tasks_ttls = [cfg.get("result_ttl", 0) for cfg in tasks_defs.values()]
+                tasks_ttls = [cfg.get("result_ttl", 0) for cfg in list(tasks_defs.values())]
 
                 result_ttl = max([default_job_timeout] + tasks_ttls)
 
@@ -92,7 +93,7 @@ def perform_action(self, action, query, destination_queue):
             # In this case we could also loose some jobs that were queued after
             # the MongoDB update. They will be "lost" and requeued later like the other case
             # after the Redis BLPOP
-            if query.keys() == ["queue"]:
+            if list(query.keys()) == ["queue"]:
                 Queue(query["queue"]).empty()
 
         elif action in ("requeue", "requeue_retry"):
@@ -135,6 +136,6 @@ def perform_action(self, action, query, destination_queue):
                     Queue(destination_queue or queue).enqueue_job_ids(
                         [str(x) for x in jobs_by_queue[queue]])
 
-        print stats
+        print(stats)
 
         return stats

mrq/bin/mrq_run.py

@@ -1,4 +1,6 @@
 #!/usr/bin/env python
+from __future__ import print_function
+
 import os
 
 # Needed to make getaddrinfo() work in pymongo on Mac OS X
@@ -40,13 +42,13 @@ def main():
         # mrq-run taskpath a 1 b 2 => {"a": "1", "b": "2"}
         for group in utils.group_iter(cfg["taskargs"], n=2):
             if len(group) != 2:
-                print "Number of arguments wasn't even"
+                print("Number of arguments wasn't even")
                 sys.exit(1)
             params[group[0]] = group[1]
 
     if cfg["queue"]:
         ret = queue_job(cfg["taskpath"], params, queue=cfg["queue"])
-        print ret
+        print(ret)
     else:
         worker_class = load_class_by_path(cfg["worker_class"])
         job = worker_class.job_class(None)
@@ -58,7 +60,7 @@ def main():
         job.datestarted = datetime.datetime.utcnow()
         set_current_job(job)
         ret = job.perform()
-        print json_stdlib.dumps(ret, cls=MongoJSONEncoder)  # pylint: disable=no-member
+        print(json_stdlib.dumps(ret, cls=MongoJSONEncoder))  # pylint: disable=no-member
 
     # This shouldn't be needed as the process will exit and close any remaining sockets
     # connections.redis.connection_pool.disconnect()

mrq/bin/mrq_worker.py

@@ -8,15 +8,19 @@
     os.environ["GEVENT_RESOLVER"] = "ares"
 
 from gevent import monkey
-monkey.patch_all()
+monkey.patch_all(subprocess=False)
 
 import sys
 import tempfile
 import signal
-import subprocess32 as subprocess
 import psutil
 import argparse
 
+try:
+    import subprocess32 as subprocess
+except:
+    import subprocess
+
 sys.path.insert(0, os.getcwd())
 
 from mrq import config

mrq/config.py

@@ -1,3 +1,4 @@
+from __future__ import print_function
 import argparse
 import os
 import sys
@@ -325,7 +326,7 @@ def add_parser_args(parser, config_type):
             '--report_file',
             default="",
             action='store',
-            type=unicode,
+            type=str,
             help='Filepath of a json dump of the worker status. Disabled if none')
 
         parser.add_argument(
@@ -402,7 +403,7 @@ def get_config(
     # line
     from_args = {}
     if "args" in sources:
-        for k, v in parser.parse_args().__dict__.iteritems():
+        for k, v in parser.parse_args().__dict__.items():
             if default_config[k] != v:
                 from_args[k] = v
 
@@ -422,7 +423,7 @@ def get_config(
         sys.path.insert(0, os.path.dirname(config_file))
         config_module = __import__(os.path.basename(config_file.replace(".py", "")))
         sys.path.pop(0)
-        for k, v in config_module.__dict__.iteritems():
+        for k, v in config_module.__dict__.items():
 
             # We only keep variables starting with an uppercase character.
             if k[0].isupper():
@@ -462,11 +463,11 @@ def print_profiling():
         atexit.register(print_profiling)
 
     if merged_config["version"]:
-        print "MRQ version: %s" % VERSION
-        print "Python version: %s" % sys.version
+        print("MRQ version: %s" % VERSION)
+        print("Python version: %s" % sys.version)
         sys.exit(1)
 
     if "no_import_patch" in from_args:
-        print "WARNING: --no_import_patch will be deprecated in MRQ 1.0!"
+        print("WARNING: --no_import_patch will be deprecated in MRQ 1.0!")
 
     return merged_config

mrq/context.py

@@ -1,7 +1,11 @@
+from future import standard_library
+standard_library.install_aliases()
+from builtins import next
+from builtins import map
 from .logger import Logger
 import gevent
 import gevent.pool
-import urlparse
+import urllib.parse
 import time
 import pymongo
 import traceback
@@ -108,12 +112,12 @@ def versiontuple(v):
         return tuple(map(int, (v.split("."))))
 
     if attr.startswith("redis"):
-        if type(config_obj) in [str, unicode]:
+        if type(config_obj) in [str]:
 
             import redis as pyredis
 
-            urlparse.uses_netloc.append('redis')
-            redis_url = urlparse.urlparse(config_obj)
+            urllib.parse.uses_netloc.append('redis')
+            redis_url = urllib.parse.urlparse(config_obj)
 
             log.info("%s: Connecting to Redis at %s..." %
                      (attr, redis_url.hostname))
@@ -124,7 +128,8 @@ def versiontuple(v):
                 db=int((redis_url.path or "").replace("/", "") or "0"),
                 password=redis_url.password,
                 max_connections=int(config.get("redis_max_connections")),
-                timeout=int(config.get("redis_timeout"))
+                timeout=int(config.get("redis_timeout")),
+                decode_responses=True
             )
             return pyredis.StrictRedis(connection_pool=redis_pool)
 
@@ -134,7 +139,7 @@ def versiontuple(v):
 
     elif attr.startswith("mongodb"):
 
-        if type(config_obj) in [str, unicode]:
+        if type(config_obj) in [str]:
 
             if attr == "mongodb_logs" and config_obj == "1":
                 return connections.mongodb_jobs
@@ -218,7 +223,7 @@ def inner_func(*args):
 
         try:
           ret = func(*args)
-        except Exception, exc:
+        except Exception as exc:
           trace = traceback.format_exc()
           log.error("Error in subpool: %s \n%s" % (exc, trace))
           raise
@@ -271,7 +276,7 @@ def inner_func(*args):
 
     try:
       ret = func(*args)
-    except Exception, exc:
+    except Exception as exc:
       trace = traceback.format_exc()
       log.error("Error in subpool: %s \n%s" % (exc, trace))
       raise

mrq/dashboard/app.py

@@ -1,3 +1,6 @@
+from __future__ import print_function
+from future import standard_library
+standard_library.install_aliases()
 from gevent import monkey
 monkey.patch_all()
 
@@ -41,7 +44,7 @@
 @requires_auth
 def root():
     return render_template("index.html", MRQ_CONFIG={
-        k: v for k, v in cfg.items() if k in WHITELISTED_MRQ_CONFIG_KEYS
+        k: v for k, v in list(cfg.items()) if k in WHITELISTED_MRQ_CONFIG_KEYS
     })
 
 
@@ -149,10 +152,10 @@ def build_api_datatables_query(req):
             try:
                 params_dict = json.loads(req.args.get("params"))
 
-                for key in params_dict.keys():
+                for key in list(params_dict.keys()):
                     query["params.%s" % key] = params_dict[key]
             except Exception as e:  # pylint: disable=broad-except
-                print "Error will converting form JSON: %s" % e
+                print("Error will converting form JSON: %s" % e)
 
     return query
 
@@ -301,7 +304,7 @@ def api_job_traceback(job_id):
 @app.route('/api/jobaction', methods=["POST"])
 @requires_auth
 def api_job_action():
-    params = {k: v for k, v in request.form.iteritems()}
+    params = {k: v for k, v in request.form.items()}
     if params.get("status") and "-" in params.get("status"):
         params["status"] = params.get("status").split("-")
     return jsonify({"job_id": queue_job("mrq.basetasks.utils.JobAction",

mrq/job.py

@@ -1,3 +1,7 @@
+from future import standard_library
+standard_library.install_aliases()
+from builtins import str
+from builtins import object
 import datetime
 from bson import ObjectId
 import time
@@ -10,12 +14,12 @@
 from collections import defaultdict
 import traceback
 import sys
-import urlparse
+import urllib.parse
 import re
 import linecache
 import fnmatch
 import encodings
-import copy_reg
+import copyreg
 from . import context
 
 
@@ -54,7 +58,10 @@ def __init__(self, job_id, queue=None, start=False, fetch=False):
         if job_id is None:
             self.id = None
         else:
-            self.id = ObjectId(job_id)
+            if isinstance(job_id, bytes):
+                self.id = ObjectId(job_id.decode('utf-8'))
+            else:
+                self.id = ObjectId(job_id)
 
         self.data = None
         self.saved = True
@@ -463,10 +470,10 @@ def set_current_io(self, io_data):
     def trace_memory_clean_caches(self):
         """ Avoid polluting results with some builtin python caches """
 
-        urlparse.clear_cache()
+        urllib.parse.clear_cache()
         re.purge()
         linecache.clearcache()
-        copy_reg.clear_extension_cache()
+        copyreg.clear_extension_cache()
 
         if hasattr(fnmatch, "purge"):
             fnmatch.purge()  # pylint: disable=no-member

mrq/logger.py

@@ -1,16 +1,27 @@
+from __future__ import print_function
+from builtins import object
 
 from collections import defaultdict
 import datetime
-
+import sys
+PY3 = sys.version_info > (3,)
 
 def _encode_if_unicode(string):
+
+    if PY3:
+        return string
+
     if isinstance(string, unicode):
         return string.encode("utf-8", "replace")
     else:
         return string
 
 
 def _decode_if_str(string):
+
+    if PY3:
+        return str(string)
+
     if isinstance(string, str):
         return string.decode("utf-8", "replace")
     else:
@@ -62,9 +73,9 @@ def log(self, level, *args, **kwargs):
 
         if not self.quiet:
             try:
-                print _encode_if_unicode(formatted)
+                print(_encode_if_unicode(formatted))
             except UnicodeDecodeError:
-                print formatted
+                print(formatted)
 
         if self.collection is False:
             return
@@ -88,10 +99,10 @@ def flush(self, w=0):
         inserts = [{
             "worker": k,
             "logs": "\n".join(v) + "\n"
-        } for k, v in self.buffer["workers"].iteritems()] + [{
+        } for k, v in self.buffer["workers"].items()] + [{
             "job": k,
             "logs": "\n".join(v) + "\n"
-        } for k, v in self.buffer["jobs"].iteritems()]
+        } for k, v in self.buffer["jobs"].items()]
 
         if len(inserts) == 0:
             return

mrq/monkey.py

@@ -1,3 +1,6 @@
+from future import standard_library
+standard_library.install_aliases()
+from past.builtins import basestring
 from .context import get_current_job, get_current_worker
 import time
 import random
@@ -339,7 +342,7 @@ def connect(old_method, self, *args, **kwargs):
 
         return ret
 
-    from httplib import HTTPConnection, HTTPSConnection
+    from http.client import HTTPConnection, HTTPSConnection
 
     patch_method(HTTPConnection, "request", request)
     patch_method(HTTPConnection, "connect", connect)
@@ -392,7 +395,7 @@ def _Cursor__send_message(self, *args, **kwargs):
                 collection = self._Cursor__collection.name  # pylint: disable=no-member
 
                 if collection == "$cmd":
-                    items = self._Cursor__spec.items()  # pylint: disable=no-member
+                    items = list(self._Cursor__spec.items())  # pylint: disable=no-member
                     if len(items) > 0:
                         subtype, collection = items[0]