I/O: Migrate from ingestr to omniload

.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

@@ -7,6 +7,7 @@
 - Dependency: Update sqlalchemy-crate version to 0.43.1
 - Removed support for Rockset
 - Fixed a bug in `cfr jobstats ui` where slider will pick an initial value too large
+- I/O: Migrated from `ingestr` to `omniload`
 
 ## 2026/06/17 v0.0.49
 - Stopped leaking password to log output in `ctk cfr jobstats collect`.

cratedb_toolkit/io/ingestr/api.py → cratedb_toolkit/io/omniload/api.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/omniload/boot.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/router.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/model.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"

doc/.gitignore

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

doc/backlog/io.md

@@ -1,10 +1,10 @@
 # Backlog for `ctk {load,save}`
 
 ## General
-- ingestr: Educate about batch size (`page_size`)
-- ingestr: More parameters for Kinesis
-  - https://getbruin.com/docs/ingestr/commands/ingest.html
-  - https://getbruin.com/docs/ingestr/supported-sources/kinesis.html
+- omniload: Educate about batch size (`page_size`)
+- omniload: More parameters for Kinesis
+  - https://omniload.readthedocs.io/commands/ingest.html
+  - https://omniload.readthedocs.io/supported-sources/kinesis.html
   - ```python
     # incremental_strategy="delete+insert",
     # mask=["temperature:noise:0.35", "temperature:round:100"],

doc/backlog/main.md

@@ -1,8 +1,6 @@
 # Main Backlog
 
 ## Iteration +0
-- Exercise data imports from AWS S3 and other Object Storage providers.
-  Evaluate the new `ingestr` subsystem.
 - A new item for the datasets API
   https://github.com/dr5hn/countries-states-cities-database
 

doc/install.md

@@ -30,7 +30,7 @@ ctk --version
 
 Alternatively, use a container image from GHCR. Use `ghcr.io/crate/cratedb-toolkit`
 for the default toolset. To use the "ingest" image (which includes the `io-ingest`
-extras and the ingestr-based I/O adapters), use `ghcr.io/crate/cratedb-toolkit-ingest`.
+extras and the omniload-based I/O adapters), use `ghcr.io/crate/cratedb-toolkit-ingest`.
 The default image does not include these extras or drivers.
 
 Run with Docker or Podman.

doc/io/_db-options.md

@@ -22,4 +22,4 @@ ctk load "protocol://?table=query:SELECT * FROM sys.summits WHERE height > 4000"
 ```
 
 
-[custom queries for SQL sources]: https://bruin-data.github.io/ingestr/supported-sources/custom_queries.html
+[custom queries for SQL sources]: https://omniload.readthedocs.io/supported-sources/custom_queries.html

doc/io/data-warehouse/databricks/index.md

@@ -152,5 +152,5 @@ See also the documentation about [SQLAlchemy's Databricks dialect].
 
 [cratedb-toolkit]: https://pypi.org/project/cratedb-toolkit/
 [Databricks]: https://www.databricks.com/
-[Databricks OAuth M2M authentication]: https://getbruin.com/docs/ingestr/supported-sources/databricks.html
+[Databricks OAuth M2M authentication]: https://omniload.readthedocs.io/supported-sources/databricks.html
 [SQLAlchemy's Databricks dialect]: https://docs.databricks.com/aws/en/dev-tools/sqlalchemy

doc/io/database/postgresql/research.md

@@ -2,7 +2,7 @@
 
 ## Full-load
 
-This subsystem is already covered by ingestr/dlt.
+This subsystem is already covered by omniload/dlt (ex. ingestr/dlt).
 
 ## CDC
 

doc/io/file/index.md

@@ -17,7 +17,7 @@ ctk load \
     "s3://?access_key_id=${AWS_ACCESS_KEY_ID}&secret_access_key=${AWS_SECRET_ACCESS_KEY}&table=openaq-fetches/realtime/2023-02-25/1677351953_eea_2aa299a7-b688-4200-864a-8df7bac3af5b.ndjson#jsonl" \
     "crate://crate:na@localhost:4200/testdrive/s3_ndjson"
 ```
-See documentation about [ingestr and Amazon S3] about details of the URI format,
+See documentation about [omniload and Amazon S3] about details of the URI format,
 file globbing patterns, compression options, and file type hinting options.
 :::{div}
 :style: font-size: small
@@ -43,4 +43,4 @@ ctk load \
 ```
 
 
-[ingestr and Amazon S3]: https://bruin-data.github.io/ingestr/supported-sources/s3.html
+[omniload and Amazon S3]: https://omniload.readthedocs.io/supported-sources/s3.html

doc/io/index.md

@@ -262,7 +262,7 @@ managed/index
 [Databricks]: https://www.databricks.com/
 [DuckDB]: https://github.com/duckdb/duckdb
 [DynamoDB]: https://aws.amazon.com/dynamodb/
-[incremental loading]: https://bruin-data.github.io/ingestr/getting-started/incremental-loading.html
+[incremental loading]: https://omniload.readthedocs.io/getting-started/incremental-loading.html
 [InfluxDB]: https://github.com/influxdata/influxdb
 [MongoDB]: https://github.com/mongodb/mongo
 [MongoDB Atlas]: https://www.mongodb.com/atlas

doc/io/search-engine/elasticsearch/index.md

@@ -54,7 +54,7 @@ wget https://cdn.crate.io/downloads/datasets/cratedb-datasets/academy/chicago-da
 ```
 Import data into Elasticsearch.
 ```shell
-ingestr ingest --yes \
+omniload ingest --yes \
   --source-uri "csv://taxi_details.csv" \
   --source-table "data" \
   --dest-uri "elasticsearch://localhost:9200?secure=false" \

doc/io/service/index.md

@@ -473,19 +473,19 @@ ctk load \
 ```
 
 
-[Anthropic entities]: https://bruin-data.github.io/ingestr/supported-sources/anthropic.html#available-tables
-[Asana entities]: https://bruin-data.github.io/ingestr/supported-sources/asana.html#tables
-[Attio entities]: https://bruin-data.github.io/ingestr/supported-sources/attio.html#tables
-[Facebook Ads entities]: https://bruin-data.github.io/ingestr/supported-sources/facebook-ads.html#tables
-[GitHub entities]: https://bruin-data.github.io/ingestr/supported-sources/github.html#tables
-[Google Ads entities]: https://bruin-data.github.io/ingestr/supported-sources/google-ads.html#tables
-[Google Analytics entities]: https://bruin-data.github.io/ingestr/supported-sources/google_analytics.html#available-tables
-[HubSpot entities]: https://bruin-data.github.io/ingestr/supported-sources/hubspot.html#tables
-[Jira entities]: https://bruin-data.github.io/ingestr/supported-sources/jira.html#tables
-[Salesforce entities]: https://bruin-data.github.io/ingestr/supported-sources/salesforce.html#tables
-[Shopify entities]: https://bruin-data.github.io/ingestr/supported-sources/shopify.html#tables
-[Slack entities]: https://bruin-data.github.io/ingestr/supported-sources/slack.html#tables
-[Stripe entities]: https://bruin-data.github.io/ingestr/supported-sources/stripe.html#all-endpoints
-[Wise entities]: https://bruin-data.github.io/ingestr/supported-sources/wise.html#tables
-[Zendesk entities]: https://bruin-data.github.io/ingestr/supported-sources/zendesk.html#tables
-[Zoom entities]: https://bruin-data.github.io/ingestr/supported-sources/zoom.html#uri-format
+[Anthropic entities]: https://omniload.readthedocs.io/supported-sources/anthropic.html#available-tables
+[Asana entities]: https://omniload.readthedocs.io/supported-sources/asana.html#tables
+[Attio entities]: https://omniload.readthedocs.io/supported-sources/attio.html#tables
+[Facebook Ads entities]: https://omniload.readthedocs.io/supported-sources/facebook-ads.html#tables
+[GitHub entities]: https://omniload.readthedocs.io/supported-sources/github.html#tables
+[Google Ads entities]: https://omniload.readthedocs.io/supported-sources/google-ads.html#tables
+[Google Analytics entities]: https://omniload.readthedocs.io/supported-sources/google_analytics.html#available-tables
+[HubSpot entities]: https://omniload.readthedocs.io/supported-sources/hubspot.html#tables
+[Jira entities]: https://omniload.readthedocs.io/supported-sources/jira.html#tables
+[Salesforce entities]: https://omniload.readthedocs.io/supported-sources/salesforce.html#tables
+[Shopify entities]: https://omniload.readthedocs.io/supported-sources/shopify.html#tables
+[Slack entities]: https://omniload.readthedocs.io/supported-sources/slack.html#tables
+[Stripe entities]: https://omniload.readthedocs.io/supported-sources/stripe.html#all-endpoints
+[Wise entities]: https://omniload.readthedocs.io/supported-sources/wise.html#tables
+[Zendesk entities]: https://omniload.readthedocs.io/supported-sources/zendesk.html#tables
+[Zoom entities]: https://omniload.readthedocs.io/supported-sources/zoom.html#uri-format