I/O: Migrate from ingestr to omniload

ingestr v1 started using the Functional Source License [1], which
prevents commercial use for two years. omniload keeps using the
MIT license.

[1] https://fsl.software/

Author: amotl

.github/workflows/postgresql.yml

@@ -5,19 +5,19 @@ on:
   pull_request:
     paths:
     - '.github/workflows/postgresql.yml'
-    - 'cratedb_toolkit/io/ingestr/**'
+    - 'cratedb_toolkit/io/omniload/**'
     - 'cratedb_toolkit/testing/testcontainers/postgresql.py'
     - 'cratedb_toolkit/testing/testcontainers/util.py'
-    - 'tests/io/ingestr/*postgresql*'
+    - 'tests/io/omniload/*postgresql*'
     - 'pyproject.toml'
   push:
     branches: [ main ]
     paths:
     - '.github/workflows/postgresql.yml'
-    - 'cratedb_toolkit/io/ingestr/**'
+    - 'cratedb_toolkit/io/omniload/**'
     - 'cratedb_toolkit/testing/testcontainers/postgresql.py'
     - 'cratedb_toolkit/testing/testcontainers/util.py'
-    - 'tests/io/ingestr/*postgresql*'
+    - 'tests/io/omniload/*postgresql*'
     - 'pyproject.toml'
 
   # Allow job to be triggered manually.

.gitignore

@@ -12,7 +12,7 @@ build
 coverage.xml
 node_modules
 /cfr
-/doc/_build
+/doc/_build/*
 /tmp
 /var
 /DOWNLOAD

CHANGES.md

@@ -4,6 +4,11 @@
 - Breaking change: Dropped support for Python 3.8 and 3.9, which have reached
   end-of-life. The minimum supported Python version is now 3.10.
 - Fixed a failing SQL statement during `ctk info cluster`
+- I/O: Migrated from `ingestr` to `omniload`. ingestr v1 started using
+  the [Functional Source License] which prevents commercial use
+  for two years. omniload keeps using the MIT license.
+
+[Functional Source License]: https://fsl.software/
 
 ## 2026/06/17 v0.0.49
 - Stopped leaking password to log output in `ctk cfr jobstats collect`.

cratedb_toolkit/io/ingestr/boot.py

@@ -3,51 +3,51 @@
 
 from yarl import URL
 
-from cratedb_toolkit.io.ingestr.boot import import_ingestr
+from cratedb_toolkit.io.omniload.boot import import_omniload
 from cratedb_toolkit.model import DatabaseAddress
 
 logger = logging.getLogger(__name__)
 
 
 @lru_cache(maxsize=1)
-def _get_ingestr():
+def _get_omniload():
     """
-    Get an ingestr API handle, cached.
+    Get an omniload API handle, cached.
     """
-    return import_ingestr()
+    return import_omniload()
 
 
-def ingestr_select(source_url: str) -> bool:
+def omniload_select(source_url: str) -> bool:
     """
-    Whether to select `ingestr` for this data source.
+    Whether to select `omniload` for this data source.
     """
-    ingestr_available, ingestr, ConfigFieldMissingException = _get_ingestr()
+    omniload_available, omniload, ConfigFieldMissingException = _get_omniload()
 
-    if not ingestr_available:
-        logger.debug("ingestr is not installed")
+    if not omniload_available:
+        logger.debug("omniload is not installed")
         return False
     try:
-        factory = ingestr.src.factory.SourceDestinationFactory(source_url, "csv:////tmp/foobar.csv")
+        factory = omniload.src.factory.SourceDestinationFactory(source_url, "csv:////tmp/foobar.csv")
         factory.get_source()
-        scheme = ingestr.src.factory.parse_scheme_from_uri(source_url)
-        logger.info(f"Selecting ingestr for source scheme: {scheme}")
+        scheme = omniload.src.factory.parse_scheme_from_uri(source_url)
+        logger.info(f"Selecting omniload for source scheme: {scheme}")
         return True
     except (ImportError, ValueError, AttributeError) as ex:
         if "Unsupported source scheme" in str(ex):
-            logger.debug(f"Failed to select ingestr for source url '{source_url}': {ex}")
+            logger.debug(f"Failed to select omniload for source url '{source_url}': {ex}")
         else:
-            logger.exception(f"Unexpected error with ingestr for source url: {source_url}")
+            logger.exception(f"Unexpected error with omniload for source url: {source_url}")
         return False
     except Exception:
-        logger.exception(f"Unexpected error with ingestr for source url: {source_url}")
+        logger.exception(f"Unexpected error with omniload for source url: {source_url}")
         return False
 
 
-def ingestr_copy(source_url: str, target_address: DatabaseAddress, progress: bool = False) -> bool:
+def omniload_copy(source_url: str, target_address: DatabaseAddress, progress: bool = False) -> bool:
     """
-    Invoke data transfer to CrateDB from any source provided by `ingestr`.
+    Invoke data transfer to CrateDB from any source provided by `omniload`.
 
-    https://cratedb-toolkit.readthedocs.io/io/ingestr/
+    https://cratedb-toolkit.readthedocs.io/io/omniload/
 
     Synopsis:
 
@@ -63,11 +63,11 @@ def ingestr_copy(source_url: str, target_address: DatabaseAddress, progress: boo
             "postgresql://pguser:secret11@postgresql.example.org:5432/postgres?table=public.diamonds" \
             "crate://crate:na@localhost:4200/testdrive/ibis_diamonds"
     """
-    ingestr_available, ingestr, ConfigFieldMissingException = _get_ingestr()
+    omniload_available, omniload, ConfigFieldMissingException = _get_omniload()
 
     # Sanity checks.
-    if not ingestr_available:
-        raise ModuleNotFoundError("ingestr subsystem not installed")
+    if not omniload_available:
+        raise ModuleNotFoundError("omniload subsystem not installed")
 
     # Compute source and target URLs and table names.
     # Table names use dotted notation `<schema>.<table>`.
@@ -91,7 +91,7 @@ def ingestr_copy(source_url: str, target_address: DatabaseAddress, progress: boo
 
     target_uri, target_table_address = target_address.decode()
     target_table = target_table_address.fullname
-    target_url = target_address.to_ingestr_url()
+    target_url = target_address.to_omniload_url()
 
     if not source_table:
         raise ValueError("Source table is required")
@@ -101,7 +101,7 @@ def ingestr_copy(source_url: str, target_address: DatabaseAddress, progress: boo
     if source_fragment:
         source_table += f"#{source_fragment}"
 
-    logger.info("Invoking ingestr")
+    logger.info("Invoking omniload")
     logger.info(f"Source URL: {source_url_obj}")
     logger.info(f"Target URL: {target_url}")
     logger.info(f"Source Table: {source_table}")
@@ -114,15 +114,14 @@ def ingestr_copy(source_url: str, target_address: DatabaseAddress, progress: boo
         dest_uri=str(target_url),
         source_table=source_table,
         dest_table=target_table,
-        yes=True,
     )
     if start_date is not None:
         ingest_kwargs["interval_start"] = start_date
     if batch_size is not None:
         ingest_kwargs["page_size"] = batch_size
 
     try:
-        ingestr.main.ingest(**ingest_kwargs)
+        omniload.main.ingest(**ingest_kwargs)
         return True
     except ConfigFieldMissingException:
         logger.error(

cratedb_toolkit/io/ingestr/__init__.py cratedb_toolkit/io/omniload/__init__.pycratedb_toolkit/io/ingestr/__init__.py renamed to cratedb_toolkit/io/omniload/__init__.py

@@ -0,0 +1,36 @@
+import logging
+
+from boltons.urlutils import URL
+
+logger = logging.getLogger(__name__)
+
+
+def import_omniload():
+    """Import omniload with CrateDB destination adapter."""
+    try:
+        import dlt_cratedb  # noqa: F401
+        import omniload.main
+        import omniload.src.factory
+        from dlt.common.configuration import ConfigFieldMissingException
+        from omniload.src.destinations import GenericSqlDestination
+
+        class CrateDBDestination(GenericSqlDestination):
+            def dlt_dest(self, uri: str, **kwargs):
+                uri = self._replace_url(uri)
+                import dlt_cratedb.impl.cratedb.factory
+
+                return dlt_cratedb.impl.cratedb.factory.cratedb(credentials=uri, **kwargs)
+
+            @staticmethod
+            def _replace_url(uri: str) -> str:
+                url_obj = URL(uri)
+                if url_obj.scheme == "cratedb":
+                    url_obj.scheme = "postgres"
+                return str(url_obj)
+
+        omniload.src.factory.SourceDestinationFactory.destinations["cratedb"] = CrateDBDestination
+
+        return True, omniload, ConfigFieldMissingException
+    except ImportError:
+        logger.error("Could not import omniload")
+        return False, None, None

cratedb_toolkit/io/ingestr/api.py cratedb_toolkit/io/omniload/api.pycratedb_toolkit/io/ingestr/api.py renamed to cratedb_toolkit/io/omniload/api.py

@@ -119,12 +119,12 @@ def load_table(
 
             return from_iceberg(str(source_url_obj), target_url)
 
-        from cratedb_toolkit.io.ingestr.api import ingestr_copy, ingestr_select
+        from cratedb_toolkit.io.omniload.api import omniload_copy, omniload_select
 
         source_url = str(source_url_obj)
 
-        if ingestr_select(source_url):
-            return ingestr_copy(source_url, target, progress=True)
+        if omniload_select(source_url):
+            return omniload_copy(source_url, target, progress=True)
 
         else:
             raise NotImplementedError(f"Importing resource not implemented yet: {source_url_obj}")

cratedb_toolkit/io/omniload/boot.py

@@ -141,9 +141,9 @@ def to_postgresql_url(self, port: int = 5432) -> URL:
         uri.port = port
         return uri
 
-    def to_ingestr_url(self, port: int = 5432) -> URL:
+    def to_omniload_url(self, port: int = 5432) -> URL:
         """
-        Return the `cratedb://` variant of the database URI, suitable for `ingestr`.
+        Return the `cratedb://` variant of the database URI, suitable for `omniload`.
         """
         uri = deepcopy(self.to_postgresql_url(port))
         uri.scheme = "cratedb"

cratedb_toolkit/io/router.py

@@ -1 +1 @@
-/_build
+_build/*