exclude non required schemas

pgsync/__init__.py

@@ -2,4 +2,4 @@
 
 __author__ = "Tolu Aina"
 __email__ = "[email protected]"
-__version__ = "3.2.0"
+__version__ = "3.2.1"

pgsync/node.py

@@ -284,6 +284,7 @@ class Tree(threading.local):
     def __post_init__(self):
         self.tables: t.Set[str] = set()
         self.__nodes: t.Dict[Node] = {}
+        self.__schemas: t.Set[str] = set()
         self.root: t.Optional[Node] = None
         self.build(self.nodes)
 
@@ -334,6 +335,7 @@ def build(self, nodes: dict) -> Node:
             node.add_child(self.build(child))
 
         self.__nodes[key] = node
+        self.__schemas.add(schema)
         return node
 
     def get_node(self, table: str, schema: str) -> Node:
@@ -352,3 +354,7 @@ def get_node(self, table: str, schema: str) -> Node:
             else:
                 raise RuntimeError(f"Node for {schema}.{table} not found")
         return self.__nodes[key]
+
+    @property
+    def schemas(self) -> t.Set[str]:
+        return self.__schemas

pgsync/sync.py

@@ -410,6 +410,11 @@ def logical_slot_changes(
                         f"Error parsing row: {e}\nRow data: {row.data}"
                     )
                     raise
+
+                # filter out unknown schemas
+                if payload.schema not in self.tree.schemas:
+                    continue
+
                 payloads.append(payload)
 
                 j: int = i + 1
@@ -811,7 +816,10 @@ def _payloads(self, payloads: t.List[Payload]) -> None:
         # e.g a through table which we need to react to.
         # in this case, we find the parent of the through
         # table and force a re-sync.
-        if payload.table not in self.tree.tables:
+        if (
+            payload.table not in self.tree.tables
+            or payload.schema not in self.tree.schemas
+        ):
             return
 
         node: Node = self.tree.get_node(payload.table, payload.schema)
@@ -1112,7 +1120,11 @@ def poll_db(self) -> None:
                 notification: t.AnyStr = conn.notifies.pop(0)
                 if notification.channel == self.database:
                     payload = json.loads(notification.payload)
-                    if payload["indices"] and self.index in payload["indices"]:
+                    if (
+                        payload["indices"]
+                        and self.index in payload["indices"]
+                        and payload["schema"] in self.tree.schemas
+                    ):
                         payloads.append(payload)
                         logger.debug(f"poll_db: {payload}")
                         self.count["db"] += 1
@@ -1134,7 +1146,11 @@ def async_poll_db(self) -> None:
             notification: t.AnyStr = self.conn.notifies.pop(0)
             if notification.channel == self.database:
                 payload = json.loads(notification.payload)
-                if payload["indices"] and self.index in payload["indices"]:
+                if (
+                    payload["indices"]
+                    and self.index in payload["indices"]
+                    and payload["schema"] in self.tree.schemas
+                ):
                     self.redis.push([payload])
                     logger.debug(f"async_poll: {payload}")
                     self.count["db"] += 1

requirements/base.txt

@@ -6,9 +6,9 @@
 #
 async-timeout==4.0.3
     # via redis
-boto3==1.34.158
+boto3==1.34.159
     # via -r requirements/base.in
-botocore==1.34.158
+botocore==1.34.159
     # via
     #   boto3
     #   s3transfer
@@ -23,7 +23,7 @@ click==8.1.7
     # via -r requirements/base.in
 elastic-transport==8.15.0
     # via elasticsearch
-elasticsearch==8.14.0
+elasticsearch==8.15.0
     # via
     #   -r requirements/base.in
     #   elasticsearch-dsl

requirements/dev.txt

@@ -8,9 +8,9 @@ async-timeout==4.0.3
     # via redis
 black==24.8.0
     # via -r requirements/dev.in
-boto3==1.34.158
+boto3==1.34.159
     # via -r requirements/base.in
-botocore==1.34.158
+botocore==1.34.159
     # via
     #   boto3
     #   s3transfer
@@ -35,7 +35,7 @@ distlib==0.3.8
     # via virtualenv
 elastic-transport==8.15.0
     # via elasticsearch
-elasticsearch==8.14.0
+elasticsearch==8.15.0
     # via
     #   -r requirements/base.in
     #   elasticsearch-dsl
@@ -47,7 +47,7 @@ events==0.5
     # via opensearch-py
 exceptiongroup==1.2.2
     # via pytest
-faker==26.3.0
+faker==27.0.0
     # via -r requirements/dev.in
 filelock==3.15.4
     # via virtualenv