more compat

Author: Windfarer

README.md

@@ -1,15 +1,14 @@
 # py-mysql-elasticsearch-sync
-Simple and fast MySQL to Elasticsearch sync tool, written in Python3.
+Simple and fast MySQL to Elasticsearch sync tool, written in Python.
 
 ## Introduction
 This tool helps you to initialize MySQL dump table to Elasticsearch by parsing mysqldump, then incremental sync MySQL table to Elasticsearch by processing MySQL Binlog.
 Also, during the binlog syncing, this tool will save the binlog sync position, so that it is easy to recover after this tool being shutdown for any reason.
 
 ## Installation
 By following these steps.
-##### 1. Python3
-This tool is written in Python3.4, so you must install Python3.4 or above first, by following [this guide](https://docs.python.org/3.4/using/index.html)
-##### 2. ibxml2 and libxslt
+
+##### 1. ibxml2 and libxslt
 Also, this tool depends on python lxml package, so that you should install  the lxml's dependecies correctly, the libxml2 and libxslt are required.
 
 For example, in CentOS:
@@ -25,15 +24,15 @@ sudo apt-get install libxml2-dev libxslt-dev python-dev
 ```
 
 See [lxml Installation](http://lxml.de/installation.html) for more infomation.
-##### 3. mysqldump
-And then, mysqldump is required.
+##### 2. mysqldump
+And then, mysqldump is required.(and enable binlog)
 
 
-##### 4. this tool
+##### 3. this tool
 Then install this tool
 
 ```
-pip3 install py-mysql-elasticsearch-sync
+pip install py-mysql-elasticsearch-sync
 ```
 
 ## Configuration
@@ -61,8 +60,6 @@ to start sync, when xml sync is over, it will also start binlog sync.
 ## Deployment
 We provide an upstart script to help you deploy this tool,since we use virtualenv for requirements isolation, you must edit it for your own condition, besides, you can deploy it in your own way.
 
-
 ## TODO
 - [ ] MultiIndex Supporting
 - [ ] Multi table Supporting
-- [ ] Python version compat

src/__init__.py

@@ -1,12 +1,19 @@
 from __future__ import print_function, unicode_literals
 from future.builtins import str, range
-try:
-    from subprocess import DEVNULL  # PY3
-except ImportError:
+import sys
+PY2 = sys.version_info[0] == 2
+
+if PY2:
     import os
     DEVNULL = open(os.devnull, 'wb')
+else:
+    from subprocess import DEVNULL
+def encode_in_py2(s):
+    if PY2:
+        return s.encode('utf-8')
+    return s
+
 import os.path
-import sys
 import yaml
 import signal
 import requests
@@ -20,7 +27,7 @@
 from pymysqlreplication import BinLogStreamReader
 from pymysqlreplication.row_event import DeleteRowsEvent, UpdateRowsEvent, WriteRowsEvent
 
-__version__ = '0.2.1'
+__version__ = '0.3.0'
 
 
 # The magic spell for removing invalid characters in xml stream.
@@ -44,7 +51,9 @@ def __init__(self):
         self.dump_cmd = 'mysqldump -h {host} -P {port} -u {user} --password={password} {db} {table} ' \
                         '--default-character-set=utf8 -X'.format(**self.config['mysql'])
 
-        self.binlog_conf = {key: self.config['mysql'][key] for key in ['host', 'port', 'user', 'password', 'db']}
+        self.binlog_conf = dict(
+            [(key, self.config['mysql'][key]) for key in ['host', 'port', 'user', 'password', 'db']]
+        )
 
         self.endpoint = 'http://{host}:{port}/{index}/{type}/_bulk'.format(
             host=self.config['elastic']['host'],
@@ -192,8 +201,9 @@ def _formatter(self, data):
                     try:
                         item['doc'][field] = serializer(item['doc'][field])
                     except ValueError as e:
-                        self.logger.error("Error occurred during format, ErrorMessage:{}, ErrorItem:{}".format(str(e),
-                                                                                                          str(item)))
+                        self.logger.error("Error occurred during format, ErrorMessage:{msg}, ErrorItem:{item}".format(
+                            msg=str(e),
+                            item=str(item)))
                         item['doc'][field] = None
             # print(item)
             yield item
@@ -204,7 +214,8 @@ def _binlog_loader(self):
         """
         if self.log_file and self.log_pos:
             resume_stream = True
-            logging.info("Resume from binlog_file: {}  binlog_pos: {}".format(self.log_file, self.log_pos))
+            logging.info("Resume from binlog_file: {file}  binlog_pos: {pos}".format(file=self.log_file,
+                                                                                     pos=self.log_pos))
         else:
             resume_stream = False
 
@@ -309,18 +320,21 @@ def _xml_parser(self, f_obj):
     def _save_binlog_record(self):
         if self.log_file and self.log_pos:
             with open(self.config['binlog_sync']['record_file'], 'w') as f:
-                logging.info("Sync binlog_file: {}  binlog_pos: {}".format(self.log_file, self.log_pos))
-                yaml.dump({"log_file": self.log_file, "log_pos": self.log_pos}, f)
+                logging.info("Sync binlog_file: {file}  binlog_pos: {pos}".format(
+                    file=self.log_file,
+                    pos=self.log_pos)
+                )
+                yaml.safe_dump({"log_file": self.log_file, "log_pos": self.log_pos}, f, default_flow_style=False)
 
     def _xml_dump_loader(self):
         mysqldump = subprocess.Popen(
-            shlex.split(self.dump_cmd),
+            shlex.split(encode_in_py2(self.dump_cmd)),
             stdout=subprocess.PIPE,
             stderr=DEVNULL,
             close_fds=True)
 
         remove_invalid_pipe = subprocess.Popen(
-            shlex.split(REMOVE_INVALID_PIPE),
+            shlex.split(encode_in_py2(REMOVE_INVALID_PIPE)),
             stdin=mysqldump.stdout,
             stdout=subprocess.PIPE,
             stderr=DEVNULL,

upstart.conf

@@ -5,9 +5,8 @@ stop on runlevel [06]
 respawn
 normal exit 0
 
-chdir <PROJECT_ROOT>
+chdir <PATH_TO_CONFIG>
 
 script
-    . venv/bin/activate
-    exec python main.py config.yaml
+    es-sync config.yaml
 end script