Include tables

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## 2.3.0
+
+- Add `include_tables` connection option to sync only the listed tables (mutually exclusive with `skip_tables`)
+
 ## 2.2.0
 
 - Add support for constraints (geodiff 2.1.0)

config.py

@@ -11,6 +11,7 @@
 import smtplib
 import subprocess
 import tempfile
+import typing
 
 from dynaconf import Dynaconf
 import dynaconf
@@ -89,6 +90,11 @@ def validate_config(config):
                 "Config error: Name of the Mergin Maps project should be provided in the namespace/name format."
             )
 
+        if "skip_tables" in conn and "include_tables" in conn:
+            raise ConfigError(
+                "Config error: `skip_tables` and `include_tables` cannot both be set for the same connection."
+            )
+
         if "skip_tables" in conn:
             if conn.skip_tables is None:
                 continue
@@ -101,7 +107,21 @@ def validate_config(config):
                 conn.skip_tables,
                 list,
             ):
-                raise ConfigError("Config error: Ignored tables parameter should be a list")
+                raise ConfigError("Config error: `skip_tables` parameter should be a list")
+
+        if "include_tables" in conn:
+            if conn.include_tables is None:
+                continue
+            elif isinstance(
+                conn.include_tables,
+                str,
+            ):
+                continue
+            elif not isinstance(
+                conn.include_tables,
+                list,
+            ):
+                raise ConfigError("Config error: `include_tables` parameter should be a list")
 
     if "notification" in config:
         settings = [
@@ -144,26 +164,35 @@ def validate_config(config):
             raise ConfigError(f"Config SMTP Error: {err}.")
 
 
+def _get_tables(tables: typing.Union[None, str, typing.List, dynaconf.vendor.box.box_list.BoxList], param_name: str):
+    if tables is None:
+        return []
+    elif isinstance(tables, str):
+        return [tables]
+    elif isinstance(tables, list):
+        if len(tables) < 1:
+            return []
+        elif isinstance(tables, dynaconf.vendor.box.box_list.BoxList):
+            return tables.to_list()
+        return tables
+    else:
+        raise ConfigError(f"Config error: `{param_name}` parameter should be a list or a string.")
+
+
 def get_ignored_tables(
     connection,
 ):
     if "skip_tables" in connection:
-        if connection.skip_tables is None:
-            return []
-        elif isinstance(
-            connection.skip_tables,
-            str,
-        ):
-            return [connection.skip_tables]
-        elif isinstance(
-            connection.skip_tables,
-            list,
-        ):
-            if len(connection.skip_tables) < 1:
-                return []
-            elif isinstance(connection.skip_tables, dynaconf.vendor.box.box_list.BoxList):
-                return connection.skip_tables.to_list()
-            return connection.skip_tables
+        return _get_tables(connection.skip_tables, "skip_tables")
+    else:
+        return []
+
+
+def get_include_tables(
+    connection,
+):
+    if "include_tables" in connection:
+        return _get_tables(connection.include_tables, "include_tables")
     else:
         return []