Filter schemas

julien-duponchelle · julien-duponchelle · commit 5a8d75f34bba · 2014-09-28T19:07:30.000+02:00
diff --git a/pymysqlreplication/binlogstream.py b/pymysqlreplication/binlogstream.py
@@ -31,7 +31,7 @@ def __init__(self, connection_settings, server_id, resume_stream=False,
                  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_tables = None, only_schemas = None):
         """
         Attributes:
             resume_stream: Start for event from position or the latest event of
@@ -43,6 +43,7 @@ def __init__(self, connection_settings, server_id, resume_stream=False,
             log_pos: Set replication start log pos
             auto_position: Use master_auto_position gtid to set position
             only_tables: An array with the tables you want to watch
+            only_schemas: An array with the schemas you want to watch
         """
         self.__connection_settings = connection_settings
         self.__connection_settings["charset"] = "utf8"
@@ -53,6 +54,7 @@ def __init__(self, connection_settings, server_id, resume_stream=False,
         self.__blocking = blocking
 
         self.__only_tables = only_tables
+        self.__only_schemas = only_schemas
         self.__allowed_events = self._allowed_event_list(only_events, ignored_events, filter_non_implemented_events)
 
         # We can't filter on packet level TABLE_MAP and rotate event because we need
@@ -238,7 +240,8 @@ def fetchone(self):
                                                self._ctl_connection,
                                                self.__use_checksum,
                                                allowed_events = self.__allowed_events_in_packet,
-                                               only_tables = self.__only_tables)
+                                               only_tables = self.__only_tables,
+                                               only_schemas = self.__only_schemas)
             if binlog_event.event_type == TABLE_MAP_EVENT:
                 self.table_map[binlog_event.event.table_id] = \
                     binlog_event.event.get_table()
diff --git a/pymysqlreplication/event.py b/pymysqlreplication/event.py
@@ -7,7 +7,9 @@
 
 
 class BinLogEvent(object):
-    def __init__(self, from_packet, event_size, table_map, ctl_connection, only_tables = None):
+    def __init__(self, from_packet, event_size, table_map, ctl_connection,
+                 only_tables = None,
+                 only_schemas = None):
         self.packet = from_packet
         self.table_map = table_map
         self.event_type = self.packet.event_type
diff --git a/pymysqlreplication/packet.py b/pymysqlreplication/packet.py
@@ -48,7 +48,10 @@ class BinLogPacketWrapper(object):
 
     }
 
-    def __init__(self, from_packet, table_map, ctl_connection, use_checksum, allowed_events = None, only_tables = None):
+    def __init__(self, from_packet, table_map, ctl_connection, use_checksum,
+                 allowed_events = None,
+                 only_tables = None,
+                 only_schemas = None):
         # -1 because we ignore the ok byte
         self.read_bytes = 0
         # Used when we want to override a value in the data buffer
@@ -86,7 +89,9 @@ def __init__(self, from_packet, table_map, ctl_connection, use_checksum, allowed
         if event_class not in allowed_events:
             return
         self.event = event_class(self, event_size_without_header, table_map,
-                                 ctl_connection, only_tables = only_tables)
+                                 ctl_connection,
+                                 only_tables = only_tables,
+                                 only_schemas = only_schemas)
         if self.event._processed == False:
             self.event = None
 
diff --git a/pymysqlreplication/row_event.py b/pymysqlreplication/row_event.py
@@ -19,6 +19,7 @@ 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.__only_schemas = kwargs["only_schemas"]
 
         #Header
         self.table_id = self._read_table_id()
@@ -31,6 +32,9 @@ 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
+        if self.__only_schemas is not None and self.schema not in self.__only_schemas:
+            self._processed = False
+            return
 
         #Event V2
         if self.event_type == BINLOG.WRITE_ROWS_EVENT_V2 or \
