add RandEvent

.github/workflows/pytest.yml

@@ -8,11 +8,8 @@ jobs:
       matrix:
         include:
           - {name: 'CPython 3.7', python: '3.7'}
-          - {name: 'CPython 3.8', python: '3.8'}
-          - {name: 'CPython 3.9', python: '3.9'}
-          - {name: 'CPython 3.10', python: '3.10'}
+          - {name: 'CPython 3.11', python: '3.11'}
           - {name: 'Pypy 3.7', python: 'pypy-3.7'}
-          - {name: 'Pypy 3.8', python: 'pypy-3.8'}
           - {name: 'Pypy 3.9', python: 'pypy-3.9'}
     name: ${{ matrix.name }}
     runs-on: ubuntu-latest

CHANGELOG

@@ -192,4 +192,7 @@
 * Fix merging error of XAPrepareEvent
 
 0.42.2 16/07/2023
-* Fix release error
+* Fix release error
+
+0.43.0 23/07/2023
+* Bump PyMySQL to 1.1.0 to solve : LookupError: unknown encoding: utf8mb3

docs/conf.py

@@ -48,9 +48,9 @@
 # built documents.
 #
 # The short X.Y version.
-version = '0.42.2'
+version = '0.43'
 # The full version, including alpha/beta/rc tags.
-release = '0.42.2'
+release = '0.43'
 
 # The language for content autogenerated by Sphinx. Refer to documentation
 # for a list of supported languages.

pymysqlreplication/binlogstream.py

@@ -14,7 +14,8 @@
     QueryEvent, RotateEvent, FormatDescriptionEvent,
     XidEvent, GtidEvent, StopEvent, XAPrepareEvent,
     BeginLoadQueryEvent, ExecuteLoadQueryEvent,
-    HeartbeatLogEvent, NotImplementedEvent, MariadbGtidEvent)
+    HeartbeatLogEvent, NotImplementedEvent, 
+    MariadbGtidEvent, RandEvent)
 from .exceptions import BinLogNotEnabled
 from .row_event import (
     UpdateRowsEvent, WriteRowsEvent, DeleteRowsEvent, TableMapEvent)
@@ -600,7 +601,8 @@ def _allowed_event_list(self, only_events, ignored_events,
                 TableMapEvent,
                 HeartbeatLogEvent,
                 NotImplementedEvent,
-                MariadbGtidEvent
+                MariadbGtidEvent,
+                RandEvent
                 ))
         if ignored_events is not None:
             for e in ignored_events:

pymysqlreplication/event.py

@@ -436,6 +436,40 @@ def _dump(self):
         print("type: %d" % (self.type))
         print("Value: %d" % (self.value))
 
+class RandEvent(BinLogEvent):
+    """
+    RandEvent is generated every time a statement uses the RAND() function.
+    Indicates the seed values to use for generating a random number with RAND() in the next statement.
+    Warning
+    - RAND_EVENT only works in statement-based logging. (need to set binlog_format as 'STATEMENT')
+    - RAND_EVENT only works when the seed number is not specified.
+
+    Attributes:
+        seed1
+        seed2
+    """
+
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
+        super(RandEvent, self).__init__(from_packet, event_size, table_map,
+                                        ctl_connection, **kwargs)
+        # Payload
+        self._seed1 = self.packet.read_uint64()
+        self._seed2 = self.packet.read_uint64()
+
+    @property
+    def seed1(self):
+        """Get the first seed value"""
+        return self._seed1
+
+    @property
+    def seed2(self):
+        """Get the second seed value"""
+        return self._seed2
+
+    def _dump(self):
+        super(RandEvent, self)._dump()
+        print("seed1: %d" % (self.seed1))
+        print("seed2: %d" % (self.seed2))
 
 class NotImplementedEvent(BinLogEvent):
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):

pymysqlreplication/packet.py

@@ -70,6 +70,7 @@ class BinLogPacketWrapper(object):
         constants.EXECUTE_LOAD_QUERY_EVENT: event.ExecuteLoadQueryEvent,
         constants.HEARTBEAT_LOG_EVENT: event.HeartbeatLogEvent,
         constants.XA_PREPARE_EVENT: event.XAPrepareEvent,
+        constants.RAND_EVENT: event.RandEvent,
         # row_event
         constants.UPDATE_ROWS_EVENT_V1: row_event.UpdateRowsEvent,
         constants.WRITE_ROWS_EVENT_V1: row_event.WriteRowsEvent,

pymysqlreplication/tests/test_basic.py

@@ -17,17 +17,17 @@
 from pymysqlreplication.constants.BINLOG import *
 from pymysqlreplication.row_event import *
 
-__all__ = ["TestBasicBinLogStreamReader", "TestMultipleRowBinLogStreamReader", "TestCTLConnectionSettings", "TestGtidBinLogStreamReader"]
+__all__ = ["TestBasicBinLogStreamReader", "TestMultipleRowBinLogStreamReader", "TestCTLConnectionSettings", "TestGtidBinLogStreamReader", "TestStatementConnectionSetting"]
 
 
 class TestBasicBinLogStreamReader(base.PyMySQLReplicationTestCase):
     def ignoredEvents(self):
         return [GtidEvent]
 
     def test_allowed_event_list(self):
-        self.assertEqual(len(self.stream._allowed_event_list(None, None, False)), 16)
-        self.assertEqual(len(self.stream._allowed_event_list(None, None, True)), 15)
-        self.assertEqual(len(self.stream._allowed_event_list(None, [RotateEvent], False)), 15)
+        self.assertEqual(len(self.stream._allowed_event_list(None, None, False)), 17)
+        self.assertEqual(len(self.stream._allowed_event_list(None, None, True)), 16)
+        self.assertEqual(len(self.stream._allowed_event_list(None, [RotateEvent], False)), 16)
         self.assertEqual(len(self.stream._allowed_event_list([RotateEvent], None, False)), 1)
 
     def test_read_query_event(self):
@@ -1002,6 +1002,37 @@ def test_parsing(self):
             gtid = Gtid("57b70f4e-20d3-11e5-a393-4a63946f7eac:1-:1")
             gtid = Gtid("57b70f4e-20d3-11e5-a393-4a63946f7eac::1")
 
+class TestStatementConnectionSetting(base.PyMySQLReplicationTestCase):
+    def setUp(self):
+        super(TestStatementConnectionSetting, self).setUp()
+        self.stream.close()
+        self.stream = BinLogStreamReader(
+            self.database,
+            server_id=1024,
+            only_events=(RandEvent, QueryEvent),
+            fail_on_table_metadata_unavailable=True
+        )
+        self.execute("SET @@binlog_format='STATEMENT'")
+
+    def test_rand_event(self):
+        self.execute("CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, data INT NOT NULL, PRIMARY KEY (id))")
+        self.execute("INSERT INTO test (data) VALUES(RAND())")
+        self.execute("COMMIT")
+
+        self.assertEqual(self.bin_log_format(), "STATEMENT")
+        self.assertIsInstance(self.stream.fetchone(), QueryEvent)
+        self.assertIsInstance(self.stream.fetchone(), QueryEvent)
+
+        expect_rand_event = self.stream.fetchone()
+        self.assertIsInstance(expect_rand_event, RandEvent)
+        self.assertEqual(type(expect_rand_event.seed1), int)
+        self.assertEqual(type(expect_rand_event.seed2), int)
+
+    def tearDown(self):
+        self.execute("SET @@binlog_format='ROW'")
+        self.assertEqual(self.bin_log_format(), "ROW")
+        super(TestStatementConnectionSetting, self).tearDown()
+
 
 if __name__ == "__main__":
     import unittest

setup.py

@@ -29,7 +29,7 @@ def run(self):
         unittest.main(tests, argv=sys.argv[:1])
 
 
-version = "0.42.2"
+version = "0.43.0"
 
 this_directory = Path(__file__).parent
 long_description = (this_directory / "README.md").read_text()
@@ -49,5 +49,5 @@ def run(self):
               "pymysqlreplication.constants",
               "pymysqlreplication.tests"],
     cmdclass={"test": TestCommand},
-    install_requires=['pymysql>=0.10'],
+    install_requires=['pymysql>=1.1.0'],
 )