Add support for ignored_tables and ignored_schema (julien-duponchelle#201)

* Add support for ignored_tables and ignored_schema

* reverting style changes

Author: abrarsheikh

pymysqlreplication/binlogstream.py

@@ -130,7 +130,8 @@ def __init__(self, connection_settings, server_id, ctl_connection_settings=None,
                  blocking=False, only_events=None, log_file=None, log_pos=None,
                  filter_non_implemented_events=True,
                  ignored_events=None, auto_position=None,
-                 only_tables=None, only_schemas=None,
+                 only_tables=None, ignored_tables=None,
+                 only_schemas=None, ignored_schemas=None,
                  freeze_schema=False, skip_to_timestamp=None,
                  report_slave=None, slave_uuid=None,
                  pymysql_wrapper=None,
@@ -148,7 +149,9 @@ def __init__(self, connection_settings, server_id, ctl_connection_settings=None,
             log_pos: Set replication start log pos (resume_stream should be true)
             auto_position: Use master_auto_position gtid to set position
             only_tables: An array with the tables you want to watch
+            ignored_tables: An array with the tables you want to skip
             only_schemas: An array with the schemas you want to watch
+            ignored_schemas: An array with the schemas you want to skip
             freeze_schema: If true do not support ALTER TABLE. It's faster.
             skip_to_timestamp: Ignore all events until reaching specified timestamp.
             report_slave: Report slave in SHOW SLAVE HOSTS.
@@ -174,7 +177,9 @@ def __init__(self, connection_settings, server_id, ctl_connection_settings=None,
             self._ctl_connection_settings.setdefault("charset", "utf8")
 
         self.__only_tables = only_tables
+        self.__ignored_tables = ignored_tables
         self.__only_schemas = only_schemas
+        self.__ignored_schemas = ignored_schemas
         self.__freeze_schema = freeze_schema
         self.__allowed_events = self._allowed_event_list(
             only_events, ignored_events, filter_non_implemented_events)
@@ -420,7 +425,9 @@ def fetchone(self):
                                                self.__use_checksum,
                                                self.__allowed_events_in_packet,
                                                self.__only_tables,
+                                               self.__ignored_tables,
                                                self.__only_schemas,
+                                               self.__ignored_schemas,
                                                self.__freeze_schema,
                                                self.__fail_on_table_metadata_unavailable)
 

pymysqlreplication/event.py

@@ -8,10 +8,12 @@
 
 class BinLogEvent(object):
     def __init__(self, from_packet, event_size, table_map, ctl_connection,
-                 only_tables = None,
-                 only_schemas = None,
-                 freeze_schema = False,
-                 fail_on_table_metadata_unavailable = False):
+                 only_tables=None,
+                 ignored_tables=None,
+                 only_schemas=None,
+                 ignored_schemas=None,
+                 freeze_schema=False,
+                 fail_on_table_metadata_unavailable=False):
         self.packet = from_packet
         self.table_map = table_map
         self.event_type = self.packet.event_type

pymysqlreplication/packet.py

@@ -88,7 +88,9 @@ class BinLogPacketWrapper(object):
     def __init__(self, from_packet, table_map, ctl_connection, use_checksum,
                  allowed_events,
                  only_tables,
+                 ignored_tables,
                  only_schemas,
+                 ignored_schemas,
                  freeze_schema,
                  fail_on_table_metadata_unavailable):
         # -1 because we ignore the ok byte
@@ -129,10 +131,12 @@ def __init__(self, from_packet, table_map, ctl_connection, use_checksum,
             return
         self.event = event_class(self, event_size_without_header, table_map,
                                  ctl_connection,
-                                 only_tables = only_tables,
-                                 only_schemas = only_schemas,
-                                 freeze_schema = freeze_schema,
-                                 fail_on_table_metadata_unavailable = fail_on_table_metadata_unavailable)
+                                 only_tables=only_tables,
+                                 ignored_tables=ignored_tables,
+                                 only_schemas=only_schemas,
+                                 ignored_schemas=ignored_schemas,
+                                 freeze_schema=freeze_schema,
+                                 fail_on_table_metadata_unavailable=fail_on_table_metadata_unavailable)
         if self.event._processed == False:
             self.event = None
 

pymysqlreplication/row_event.py

@@ -22,7 +22,9 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
                                         ctl_connection, **kwargs)
         self.__rows = None
         self.__only_tables = kwargs["only_tables"]
+        self.__ignored_tables = kwargs["ignored_tables"]
         self.__only_schemas = kwargs["only_schemas"]
+        self.__ignored_schemas = kwargs["ignored_schemas"]
 
         #Header
         self.table_id = self._read_table_id()
@@ -39,9 +41,16 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
         if self.__only_tables is not None and self.table not in self.__only_tables:
             self._processed = False
             return
+        elif self.__ignored_tables is not None and self.table in self.__ignored_tables:
+            self._processed = False
+            return
+
         if self.__only_schemas is not None and self.schema not in self.__only_schemas:
             self._processed = False
             return
+        elif self.__ignored_schemas is not None and self.schema in self.__ignored_schemas:
+            self._processed = False
+            return
 
 
         #Event V2
@@ -527,7 +536,9 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
         super(TableMapEvent, self).__init__(from_packet, event_size,
                                             table_map, ctl_connection, **kwargs)
         self.__only_tables = kwargs["only_tables"]
+        self.__ignored_tables = kwargs["ignored_tables"]
         self.__only_schemas = kwargs["only_schemas"]
+        self.__ignored_schemas = kwargs["ignored_schemas"]
         self.__freeze_schema = kwargs["freeze_schema"]
 
         # Post-Header
@@ -549,9 +560,16 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
         if self.__only_tables is not None and self.table not in self.__only_tables:
             self._processed = False
             return
+        elif self.__ignored_tables is not None and self.table in self.__ignored_tables:
+            self._processed = False
+            return
+
         if self.__only_schemas is not None and self.schema not in self.__only_schemas:
             self._processed = False
             return
+        elif self.__ignored_schemas is not None and self.schema in self.__ignored_schemas:
+            self._processed = False
+            return
 
         self.packet.advance(1)
         self.column_count = self.packet.read_length_coded_binary()

pymysqlreplication/tests/test_basic.py

@@ -148,14 +148,15 @@ def test_filtering_ignore_events(self):
         event = self.stream.fetchone()
         self.assertIsInstance(event, RotateEvent)
 
-    def test_filtering_table_event(self):
+    def test_filtering_table_event_with_only_tables(self):
         self.stream.close()
         self.assertEqual(self.bin_log_format(), "ROW")
         self.stream = BinLogStreamReader(
             self.database,
             server_id=1024,
             only_events=[WriteRowsEvent],
-            only_tables = ["test_2"])
+            only_tables = ["test_2"]
+        )
 
         query = "CREATE TABLE test_2 (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
         self.execute(query)
@@ -171,6 +172,51 @@ def test_filtering_table_event(self):
         event = self.stream.fetchone()
         self.assertEqual(event.table, "test_2")
 
+    def test_filtering_table_event_with_ignored_tables(self):
+        self.stream.close()
+        self.assertEqual(self.bin_log_format(), "ROW")
+        self.stream = BinLogStreamReader(
+            self.database,
+            server_id=1024,
+            only_events=[WriteRowsEvent],
+            ignored_tables = ["test_2"]
+        )
+
+        query = "CREATE TABLE test_2 (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
+        self.execute(query)
+        query = "CREATE TABLE test_3 (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
+        self.execute(query)
+
+        self.execute("INSERT INTO test_2 (data) VALUES ('alpha')")
+        self.execute("INSERT INTO test_3 (data) VALUES ('alpha')")
+        self.execute("INSERT INTO test_2 (data) VALUES ('beta')")
+        self.execute("COMMIT")
+        event = self.stream.fetchone()
+        self.assertEqual(event.table, "test_3")
+
+    def test_filtering_table_event_with_only_tables_and_ignored_tables(self):
+        self.stream.close()
+        self.assertEqual(self.bin_log_format(), "ROW")
+        self.stream = BinLogStreamReader(
+            self.database,
+            server_id=1024,
+            only_events=[WriteRowsEvent],
+            only_tables = ["test_2"],
+            ignored_tables = ["test_3"]
+        )
+
+        query = "CREATE TABLE test_2 (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
+        self.execute(query)
+        query = "CREATE TABLE test_3 (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
+        self.execute(query)
+
+        self.execute("INSERT INTO test_2 (data) VALUES ('alpha')")
+        self.execute("INSERT INTO test_3 (data) VALUES ('alpha')")
+        self.execute("INSERT INTO test_2 (data) VALUES ('beta')")
+        self.execute("COMMIT")
+        event = self.stream.fetchone()
+        self.assertEqual(event.table, "test_2")
+
     def test_write_row_event(self):
         query = "CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
         self.execute(query)