Support PyMysql with a version greater than 0.9.3 (#328)

Author: garenchan

pymysqlreplication/binlogstream.py

@@ -2,6 +2,7 @@
 
 import pymysql
 import struct
+from distutils.version import LooseVersion
 
 from pymysql.constants.COMMAND import COM_BINLOG_DUMP, COM_REGISTER_SLAVE
 from pymysql.cursors import DictCursor
@@ -259,7 +260,7 @@ def _register_slave(self):
 
         packet = self.report_slave.encoded(self.__server_id)
 
-        if pymysql.__version__ < "0.6":
+        if pymysql.__version__ < LooseVersion("0.6"):
             self._stream_connection.wfile.write(packet)
             self._stream_connection.wfile.flush()
             self._stream_connection.read_packet()
@@ -407,7 +408,7 @@ def __connect_to_stream(self):
             # encoded_data
             prelude += gtid_set.encoded()
 
-        if pymysql.__version__ < "0.6":
+        if pymysql.__version__ < LooseVersion("0.6"):
             self._stream_connection.wfile.write(prelude)
             self._stream_connection.wfile.flush()
         else:
@@ -424,7 +425,7 @@ def fetchone(self):
                 self.__connect_to_ctl()
 
             try:
-                if pymysql.__version__ < "0.6":
+                if pymysql.__version__ < LooseVersion("0.6"):
                     pkt = self._stream_connection.read_packet()
                 else:
                     pkt = self._stream_connection._read_packet()

pymysqlreplication/row_event.py

@@ -6,7 +6,7 @@
 import json
 
 from pymysql.util import byte2int
-from pymysql.charset import charset_to_encoding
+from pymysql.charset import charset_by_name
 
 from .event import BinLogEvent
 from .exceptions import TableMetadataUnavailableError
@@ -212,10 +212,16 @@ def __read_fsp(self, column):
             return microsecond * (10 ** (6-column.fsp))
         return 0
 
+    @staticmethod
+    def charset_to_encoding(name):
+        charset = charset_by_name(name)
+        return charset.encoding if charset else name
+
     def __read_string(self, size, column):
         string = self.packet.read_length_coded_pascal_string(size)
         if column.character_set_name is not None:
-            string = string.decode(charset_to_encoding(column.character_set_name))
+            encoding = self.charset_to_encoding(column.character_set_name)
+            string = string.decode(encoding)
         return string
 
     def __read_bit(self, column):

setup.py

@@ -50,5 +50,5 @@ def run(self):
               "pymysqlreplication.tests"],
     cmdclass={"test": TestCommand},
     extras_require={'test': tests_require},
-    install_requires=['pymysql'],
+    install_requires=['pymysql>=0.6'],
 )