MongoDB: Improve dispatching of server- vs. file-based processing

crate · Sep 12, 2024 · fb70a36 · fb70a36
1 parent b35440f
commit fb70a36
Showing 1 changed file with 2 additions and 21 deletions.
diff --git a/cratedb_toolkit/api/main.py b/cratedb_toolkit/api/main.py
@@ -125,24 +125,6 @@ def load_table(self, resource: InputOutputResource, target: TableAddress, transf
                 logger.error("Data loading failed or incomplete")
                 return False
 
-        elif source_url_obj.scheme.startswith("file"):
-            if "+bson" in source_url_obj.scheme or "+mongodb" in source_url_obj.scheme:
-                return mongodb_copy_generic(
-                    str(source_url_obj),
-                    target_url,
-                    transformation=transformation,
-                    progress=True,
-                )
-
-        elif source_url_obj.scheme.startswith("http"):
-            if "+bson" in source_url_obj.scheme or "+mongodb" in source_url_obj.scheme:
-                return mongodb_copy_generic(
-                    str(source_url_obj),
-                    target_url,
-                    transformation=transformation,
-                    progress=True,
-                )
-
         elif source_url_obj.scheme.startswith("influxdb"):
             from cratedb_toolkit.io.influxdb import influxdb_copy
 
@@ -156,7 +138,7 @@ def load_table(self, resource: InputOutputResource, target: TableAddress, transf
                 logger.error("Data loading failed or incomplete")
                 return False
 
-        elif source_url_obj.scheme.startswith("mongodb"):
+        elif source_url_obj.scheme in ["file+bson", "http+bson", "https+bson", "mongodb", "mongodb+srv"]:
             if "+cdc" in source_url_obj.scheme:
                 source_url_obj.scheme = source_url_obj.scheme.replace("+cdc", "")
                 from cratedb_toolkit.io.mongodb.api import mongodb_relay_cdc
@@ -169,11 +151,10 @@ def load_table(self, resource: InputOutputResource, target: TableAddress, transf
                     transformation=transformation,
                     progress=True,
                 )
+
         else:
             raise NotImplementedError("Importing resource not implemented yet")
 
-        return False
-
 
 def mongodb_copy_generic(
     source_url: str, target_url: str, transformation: t.Union[Path, None] = None, progress: bool = False