Add support for TINYINT(1) to bool() mapping

Author: bjoernhaeuser

pymysqlreplication/binlogstream.py

@@ -12,11 +12,11 @@ class BinLogStreamReader(object):
  '''Connect to replication stream and read event'''
 
  def __init__(self, connection_settings={}, resume_stream=False, blocking=False, only_events=None, server_id=255):
- '''
+ """
  resume_stream: Start for latest event of binlog or from older available event
  blocking: Read on stream is blocking
  only_events: Array of allowed events
- '''
+ """
  self.__connection_settings = connection_settings
  self.__connection_settings['charset'] = 'utf8'
 

pymysqlreplication/column.py

@@ -2,8 +2,9 @@
 from .constants import FIELD_TYPE
 from pymysql.util import byte2int, int2byte
 
+
 class Column(object):
- '''Definition of a column'''
+ """Definition of a column"""
 
  def __init__(self, column_type, column_schema, packet):
  self.type = column_type
@@ -12,6 +13,7 @@ def __init__(self, column_type, column_schema, packet):
  self.character_set_name = column_schema["CHARACTER_SET_NAME"]
  self.comment = column_schema["COLUMN_COMMENT"]
  self.unsigned = False
+ self.type_is_bool = False
 
  if column_schema["COLUMN_TYPE"].find("unsigned") != -1:
  self.unsigned = True
@@ -41,10 +43,11 @@ def __init__(self, column_type, column_schema, packet):
  self.fsp = packet.read_uint8()
  elif self.type == FIELD_TYPE.TIME2:
  self.fsp = packet.read_uint8()
-
+ elif self.type == FIELD_TYPE.TINY and column_schema["COLUMN_TYPE"] == "tinyint(1)":
+ self.type_is_bool = True
 
  def __read_string_metadata(self, packet, column_schema):
- metadata  = (packet.read_uint8() << 8) + packet.read_uint8()
+ metadata = (packet.read_uint8() << 8) + packet.read_uint8()
  real_type = metadata >> 8
  if real_type == FIELD_TYPE.SET or real_type == FIELD_TYPE.ENUM:
  self.type = real_type

pymysqlreplication/packet.py

@@ -107,15 +107,15 @@ def read_length_coded_binary(self):
  """
  c = byte2int(self.read(1))
  if c == NULL_COLUMN:
- return None
+  return None
  if c < UNSIGNED_CHAR_COLUMN:
- return c
+  return c
  elif c == UNSIGNED_SHORT_COLUMN:
  return self.unpack_uint16(self.read(UNSIGNED_SHORT_LENGTH))
  elif c == UNSIGNED_INT24_COLUMN:
- return self.unpack_int24(self.read(UNSIGNED_INT24_LENGTH))
+  return self.unpack_int24(self.read(UNSIGNED_INT24_LENGTH))
  elif c == UNSIGNED_INT64_COLUMN:
- return self.unpack_int64(self.read(UNSIGNED_INT64_LENGTH))
+  return self.unpack_int64(self.read(UNSIGNED_INT64_LENGTH))
 
  def read_length_coded_string(self):
  """Read a 'Length Coded String' from the data buffer.

pymysqlreplication/row_event.py

@@ -8,6 +8,7 @@
 from .constants import BINLOG
 from .column import Column
 
+
 class RowsEvent(BinLogEvent):
  def __init__(self, from_packet, event_size, table_map, ctl_connection):
  super(RowsEvent, self).__init__(from_packet, event_size, table_map, ctl_connection)
@@ -19,8 +20,8 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection):
 
  #Event V2
  if self.event_type == BINLOG.WRITE_ROWS_EVENT or \
- self.event_type == BINLOG.DELETE_ROWS_EVENT or \
- self.event_type == BINLOG.UPDATE_ROWS_EVENT:
+  self.event_type == BINLOG.DELETE_ROWS_EVENT or \
+  self.event_type == BINLOG.UPDATE_ROWS_EVENT:
  self.extra_data_length = struct.unpack('<H', self.packet.read(2))[0]
  self.extra_data = self.packet.read(self.extra_data_length / 8)
 
@@ -54,6 +55,9 @@ def _read_column_data(self, null_bitmap):
  values[name] = struct.unpack("<B", self.packet.read(1))[0]
  else:
  values[name] = struct.unpack("<b", self.packet.read(1))[0]
+
+ if column.type_is_bool:
+ values[name] = bool(values[name])
  elif column.type == FIELD_TYPE.SHORT:
  if unsigned:
  values[name] = struct.unpack("<H", self.packet.read(2))[0]
@@ -94,10 +98,11 @@ def _read_column_data(self, null_bitmap):
  # For new date format: http://dev.mysql.com/doc/internals/en/date-and-time-data-type-representation.html
  elif column.type == FIELD_TYPE.DATETIME2:
  values[name] = self.__read_datetime2(column)
- elif  column.type == FIELD_TYPE.TIME2:
+ elif column.type == FIELD_TYPE.TIME2:
  values[name] = self.__read_time2(column)
  elif column.type == FIELD_TYPE.TIMESTAMP2:
- values[name] = self.__add_fsp_to_time(datetime.datetime.fromtimestamp(self.packet.read_int_be_by_size(4)), column)
+ values[name] = self.__add_fsp_to_time(
+ datetime.datetime.fromtimestamp(self.packet.read_int_be_by_size(4)), column)
  elif column.type == FIELD_TYPE.LONGLONG:
  if unsigned:
  values[name] = self.packet.read_uint64()
@@ -110,7 +115,7 @@ def _read_column_data(self, null_bitmap):
  elif column.type == FIELD_TYPE.SET:
  #We read set columns as a bitmap telling us which options are enabled
  bit_mask = self.packet.read_uint_by_size(column.size)
- values[name] = {val for idx,val in enumerate(column.set_values) if bit_mask & 2**idx} or None
+ values[name] = {val for idx, val in enumerate(column.set_values) if bit_mask & 2 ** idx} or None
 
  elif column.type == FIELD_TYPE.BIT:
  values[name] = self.__read_bit(column)
@@ -134,9 +139,9 @@ def __add_fsp_to_time(self, time, column):
  if read > 0:
  microsecond = self.packet.read_int_be_by_size(read)
  if column.fsp % 2:
- time = time.replace(microsecond = int(microsecond / 10))
+ time = time.replace(microsecond=int(microsecond / 10))
  else:
- time = time.replace(microsecond = microsecond)
+ time = time.replace(microsecond=microsecond)
  return time
 
  def __read_string(self, size, column):
@@ -171,9 +176,9 @@ def __read_bit(self, column):
  def __read_time(self):
  time = self.packet.read_uint24()
  date = datetime.time(
- hour = int(time / 10000),
- minute = int((time % 10000) / 100),
- second = int(time % 100))
+ hour=int(time / 10000),
+ minute=int((time % 10000) / 100),
+ second=int(time % 100))
  return date
 
  def __read_time2(self, column):
@@ -187,9 +192,9 @@ def __read_time2(self, column):
 24 bits = 3 bytes'''
  data = self.packet.read_int_be_by_size(3)
  t = datetime.time(
- hour = self.__read_binary_slice(data, 2, 10, 24),
- minute = self.__read_binary_slice(data, 12, 6, 24),
- second = self.__read_binary_slice(data, 18, 6, 24))
+ hour=self.__read_binary_slice(data, 2, 10, 24),
+ minute=self.__read_binary_slice(data, 12, 6, 24),
+ second=self.__read_binary_slice(data, 18, 6, 24))
  return self.__add_fsp_to_time(t, column)
 
  def __read_date(self):
@@ -198,9 +203,9 @@ def __read_date(self):
  return None
 
  date = datetime.date(
- year = (time & ((1 << 15) - 1) << 9) >> 9,
- month = (time & ((1 << 4) - 1) << 5) >> 5,
- day = (time & ((1 << 5) - 1))
+ year=(time & ((1 << 15) - 1) << 9) >> 9,
+ month=(time & ((1 << 4) - 1) << 5) >> 5,
+ day=(time & ((1 << 5) - 1))
  )
  return date
 
@@ -219,12 +224,12 @@ def __read_datetime(self):
  return None
 
  date = datetime.datetime(
- year = year,
- month = month,
- day = day,
- hour = int(time / 10000),
- minute = int((time % 10000) / 100),
- second = int(time % 100))
+ year=year,
+ month=month,
+ day=day,
+ hour=int(time / 10000),
+ minute=int((time % 10000) / 100),
+ second=int(time % 100))
  return date
 
  def __read_datetime2(self, column):
@@ -241,12 +246,12 @@ def __read_datetime2(self, column):
  year_month = self.__read_binary_slice(data, 1, 17, 40)
  try:
  t = datetime.datetime(
- year = int(year_month / 13),
- month = year_month % 13,
- day = self.__read_binary_slice(data, 18, 5, 40),
- hour = self.__read_binary_slice(data, 23, 5, 40),
- minute = self.__read_binary_slice(data, 28, 6, 40),
- second = self.__read_binary_slice(data, 34, 6, 40))
+ year=int(year_month / 13),
+ month=year_month % 13,
+ day=self.__read_binary_slice(data, 18, 5, 40),
+ hour=self.__read_binary_slice(data, 23, 5, 40),
+ minute=self.__read_binary_slice(data, 28, 6, 40),
+ second=self.__read_binary_slice(data, 34, 6, 40))
  except ValueError:
  return None
  return self.__add_fsp_to_time(t, column)
@@ -278,7 +283,6 @@ def __read_new_decimal(self, column):
  res = "-"
  self.packet.unread(struct.pack('<B', value ^ 0x80))
 
-
  size = compressed_bytes[comp_integral]
  if size > 0:
  value = self.packet.read_int_be_by_size(size) ^ mask
@@ -316,7 +320,7 @@ def __read_binary_slice(self, binary, start, size, data_length):
  def _dump(self):
  super(RowsEvent, self)._dump()
  print("Table: %s.%s" % (self.schema, self.table))
- print("Affected columns: %d" % (self.number_of_columns))
+ print("Affected columns: %d" % self.number_of_columns)
  print("Changed rows: %d" % (len(self.rows)))
 
  def _fetch_rows(self):
@@ -332,7 +336,8 @@ def __getattr__(self, name):
 
 
 class DeleteRowsEvent(RowsEvent):
- '''This evenement is trigger when a row in database is removed'''
+ """This event is trigger when a row in the database is removed"""
+
  def __init__(self, from_packet, event_size, table_map, ctl_connection):
  super(DeleteRowsEvent, self).__init__(from_packet, event_size, table_map, ctl_connection)
  self.columns_present_bitmap = self.packet.read((self.number_of_columns + 7) / 8)
@@ -354,7 +359,8 @@ def _dump(self):
 
 
 class WriteRowsEvent(RowsEvent):
- '''This evenement is trigger when a row in database is added'''
+ """This event is triggered when a row in database is added"""
+
  def __init__(self, from_packet, event_size, table_map, ctl_connection):
  super(WriteRowsEvent, self).__init__(from_packet, event_size, table_map, ctl_connection)
  self.columns_present_bitmap = self.packet.read((self.number_of_columns + 7) / 8)
@@ -376,9 +382,10 @@ def _dump(self):
 
 
 class UpdateRowsEvent(RowsEvent):
- '''This evenement is trigger when a row in database change'''
+ """This event is triggered when a row in the database is changed"""
+
  def __init__(self, from_packet, event_size, table_map, ctl_connection):
- super(UpdateRowsEvent,self).__init__(from_packet, event_size, table_map, ctl_connection)
+ super(UpdateRowsEvent, self).__init__(from_packet, event_size, table_map, ctl_connection)
  #Body
  self.columns_present_bitmap = self.packet.read((self.number_of_columns + 7) / 8)
  self.columns_present_bitmap2 = self.packet.read((self.number_of_columns + 7) / 8)
@@ -395,7 +402,7 @@ def _fetch_one_row(self):
 
  def _dump(self):
  super(UpdateRowsEvent, self)._dump()
- print("Affected columns: %d" % (self.number_of_columns))
+ print("Affected columns: %d" % self.number_of_columns)
  print("Values:")
  for row in self.rows:
  print("--")
@@ -407,19 +414,19 @@ class TableMapEvent(BinLogEvent):
  '''This evenement describe the structure of a table.
  It's send before a change append on a table.
  A end user of the lib should have no usage of this'''
+
  def __init__(self, from_packet, event_size, table_map, ctl_connection):
  super(TableMapEvent, self).__init__(from_packet, event_size, table_map, ctl_connection)
 
  # Post-Header
  self.table_id = self._read_table_id()
  self.flags = struct.unpack('<H', self.packet.read(2))[0]
 
-
  # Payload
- self.schema_length =  byte2int(self.packet.read(1))
+ self.schema_length = byte2int(self.packet.read(1))
  self.schema = self.packet.read(self.schema_length).decode()
  self.packet.advance(1)
- self.table_length =  byte2int(self.packet.read(1))
+ self.table_length = byte2int(self.packet.read(1))
  self.table = self.packet.read(self.table_length).decode()
  self.packet.advance(1)
  self.column_count = self.packet.read_length_coded_binary()
@@ -429,7 +436,7 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection):
  if self.table_id in table_map:
  self.column_schemas = table_map[self.table_id].column_schemas
  else:
- self.column_schemas = self.__get_table_informations(self.schema, self.table)
+ self.column_schemas = self.__get_table_information(self.schema, self.table)
 
  #Read columns meta data
  column_types = list(self.packet.read(self.column_count))
@@ -444,7 +451,7 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection):
  # TODO: get this informations instead of trashing data
  # n NULL-bitmask, length: (column-length * 8) / 7
 
- def __get_table_informations(self, schema, table):
+ def __get_table_information(self, schema, table):
  cur = self._ctl_connection.cursor()
  cur.execute("""SELECT * FROM columns WHERE table_schema = %s AND table_name = %s""", (schema, table))
  return cur.fetchall()

pymysqlreplication/tests/base.py

@@ -3,14 +3,15 @@
 import copy
 from pymysqlreplication import BinLogStreamReader
 
+
 class PyMySQLReplicationTestCase(unittest.TestCase):
  '''Test the module. Be carefull it will reset your MySQL server'''
- database =  {"host":"localhost",
- "user":"root",
- "passwd":"",
- "use_unicode": True,
- "charset": "utf8",
- "db": "pymysqlreplication_test"
+ database = {"host": "localhost",
+  "user": "root",
+  "passwd": "",
+  "use_unicode": True,
+  "charset": "utf8",
+  "db": "pymysqlreplication_test"
  }
 
  def setUp(self):
@@ -57,5 +58,5 @@ def resetBinLog(self):
  self.execute("RESET MASTER")
  if self.stream is not None:
  self.stream.close()
- self.stream = BinLogStreamReader(connection_settings = self.database)
+ self.stream = BinLogStreamReader(connection_settings=self.database)
 

pymysqlreplication/tests/test_data_type.py

@@ -109,6 +109,30 @@ def test_tiny(self):
  self.assertEqual(event.rows[0]["values"]["id"], 255)
  self.assertEqual(event.rows[0]["values"]["test"], -128)
 
+ def test_tiny_maps_to_boolean_true(self):
+ create_query = "CREATE TABLE test (id TINYINT UNSIGNED NOT NULL, test BOOLEAN)"
+ insert_query = "INSERT INTO test VALUES(1, TRUE)"
+ event = self.create_and_insert_value(create_query, insert_query)
+ self.assertEqual(event.rows[0]["values"]["id"], 1)
+ self.assertEqual(type(event.rows[0]["values"]["test"]), type(True))
+ self.assertEqual(event.rows[0]["values"]["test"], True)
+
+ def test_tiny_maps_to_boolean_false(self):
+ create_query = "CREATE TABLE test (id TINYINT UNSIGNED NOT NULL, test BOOLEAN)"
+ insert_query = "INSERT INTO test VALUES(1, FALSE)"
+ event = self.create_and_insert_value(create_query, insert_query)
+ self.assertEqual(event.rows[0]["values"]["id"], 1)
+ self.assertEqual(type(event.rows[0]["values"]["test"]), type(True))
+ self.assertEqual(event.rows[0]["values"]["test"], False)
+
+ def test_tiny_maps_to_none(self):
+ create_query = "CREATE TABLE test (id TINYINT UNSIGNED NOT NULL, test BOOLEAN)"
+ insert_query = "INSERT INTO test VALUES(1, NULL)"
+ event = self.create_and_insert_value(create_query, insert_query)
+ self.assertEqual(event.rows[0]["values"]["id"], 1)
+ self.assertEqual(type(event.rows[0]["values"]["test"]), type(None))
+ self.assertEqual(event.rows[0]["values"]["test"], None)
+
  def test_short(self):
  create_query = "CREATE TABLE test (id SMALLINT UNSIGNED NOT NULL, test SMALLINT)"
  insert_query = "INSERT INTO test VALUES(65535, -32768)"