I/O: Add CSV file import, with transformations

Author: amotl

CHANGES.md

@@ -5,6 +5,7 @@
 - Kinesis: Added `ctk kinesis` CLI group with `list-checkpoints` and
   `prune-checkpoints` commands for checkpoint table maintenance
 - Dependencies: Permitted installation of click 8.3
+- I/O: Added CSV file import, with transformations
 
 ## 2026/03/16 v0.0.46
 - I/O: API improvements: `ctk {load,save} table` became `ctk {load,save}`

cratedb_toolkit/cluster/core.py

@@ -545,10 +545,10 @@ def load_table(
         ctk load kinesis+dms:///arn:aws:kinesis:eu-central-1:831394476016:stream/testdrive
         ctk load kinesis+dms:///path/to/dms-over-kinesis.jsonl
         """
-
-        self._load_table_result = self._router.load_table(
-            source=source, target=self.address, transformation=transformation
-        )
+        address = self.address
+        if target:
+            address = address.with_table_address(target)
+        self._load_table_result = self._router.load_table(source=source, target=address, transformation=transformation)
         return self
 
     def save_table(

cratedb_toolkit/io/file/__init__.py

@@ -0,0 +1,125 @@
+"""
+CSV file integration for CrateDB Toolkit.
+
+This module provides functionality to transfer data between CSV files
+and CrateDB database tables, supporting both import and export operations.
+"""
+
+import dataclasses
+import logging
+from typing import Dict, List, Optional
+
+import polars as pl
+from boltons.urlutils import URL
+
+from cratedb_toolkit.io.util import parse_uri, polars_to_cratedb
+
+logger = logging.getLogger(__name__)
+
+
+DEFAULT_SEPARATOR = ","
+DEFAULT_QUOTE_CHAR = '"'
+DEFAULT_BATCH_SIZE = 75_000
+
+
+@dataclasses.dataclass
+class CsvFileAddress:
+    """
+    Represent a CSV file location and provide loader methods.
+    """
+
+    url: URL
+    location: str
+    pipeline: Optional[List[str]] = dataclasses.field(default_factory=list)
+    batch_size: Optional[int] = DEFAULT_BATCH_SIZE
+    # TODO: What about other parameters? See `polars.io.csv.functions`.
+    separator: Optional[str] = DEFAULT_SEPARATOR
+    quote_char: Optional[str] = DEFAULT_QUOTE_CHAR
+
+    @classmethod
+    def from_url(cls, url: str) -> "CsvFileAddress":
+        """
+        Parse a CSV file location and return a CsvFileAddress object.
+
+        Examples:
+
+        csv://./var/lib/example.csv
+        https://guided-path.s3.us-east-1.amazonaws.com/demo_climate_data_export.csv
+        """
+        url_obj, location = parse_uri(url, "csv")
+        return cls(
+            url=url_obj,
+            location=location,
+            pipeline=url_obj.query_params.getlist("pipe"),
+            batch_size=int(url_obj.query_params.get("batch-size", DEFAULT_BATCH_SIZE)),
+            separator=url_obj.query_params.get("separator", DEFAULT_SEPARATOR),
+            quote_char=url_obj.query_params.get("quote-char", DEFAULT_QUOTE_CHAR),
+        )
+
+    @property
+    def storage_options(self) -> Dict[str, str]:
+        """
+        Provide file storage options.
+
+        TODO: Generalize.
+        """
+        prefixes = ["aws_", "azure_", "google_", "delta_"]
+        return self.collect_properties(self.url.query_params, prefixes)
+
+    @staticmethod
+    def collect_properties(query_params: Dict, prefixes: List) -> Dict[str, str]:
+        """
+        Collect parameters from URL query string.
+
+        TODO: Generalize.
+        """
+        opts = {}
+        for name, value in query_params.items():
+            for prefix in prefixes:
+                if name.lower().startswith(prefix) and value is not None:
+                    opts[name.upper()] = value
+                    break
+        return opts
+
+    def load_table(self) -> pl.LazyFrame:
+        """
+        Load the CSV file as a Polars LazyFrame.
+        """
+
+        # Read from data source.
+        lf = pl.scan_csv(
+            self.location,
+            separator=self.separator,
+            quote_char=self.quote_char,
+            storage_options=self.storage_options,
+        )
+
+        # Optionally apply transformations.
+        if self.pipeline:
+            from macropipe import MacroPipe
+
+            mp = MacroPipe.from_recipes(*self.pipeline)
+            lf = mp.apply(lf)
+
+        return lf
+
+
+def from_csv(source_url, target_url, progress: bool = False) -> bool:
+    """
+    Scan a CSV file from local filesystem or object store, and load into CrateDB.
+    Documentation: https://cratedb-toolkit.readthedocs.io/io/file/csv.html
+
+    See also: https://docs.pola.rs/api/python/stable/reference/api/polars.scan_csv.html
+
+    # Synopsis: Load from filesystem.
+    ctk load \
+        "csv://./var/lib/example.csv" \
+        "crate://crate@localhost:4200/demo/example"
+    """
+    source = CsvFileAddress.from_url(source_url)
+    logger.info(f"File address: {source.location}")
+    return polars_to_cratedb(
+        frame=source.load_table(),
+        target_url=target_url,
+        chunk_size=source.batch_size,
+    )

cratedb_toolkit/io/file/csv.py

@@ -107,6 +107,19 @@ def load_table(
                     progress=True,
                 )
 
+        elif (
+            source_url_obj.scheme.startswith("csv")
+            or source_url_obj.scheme.endswith("csv")
+            or source_url_obj.path.endswith(".csv")
+        ):
+            from cratedb_toolkit.io.file.csv import from_csv
+
+            adjusted_url = str(source_url_obj)
+            if source_url_obj.scheme.startswith("csv"):
+                adjusted_url = str(source_url_obj.path)
+
+            return from_csv(adjusted_url, target_url)
+
         elif source_url_obj.scheme.startswith("deltalake") or source_url_obj.scheme.endswith("deltalake"):
             from cratedb_toolkit.io.deltalake import from_deltalake
 

cratedb_toolkit/io/router.py

@@ -40,9 +40,10 @@ def polars_to_cratedb(frame: pl.LazyFrame, target_url, chunk_size: int) -> bool:
     """
     Write a Polars LazyFrame to a CrateDB table, in batches/chunks.
     """
-    cratedb_address = DatabaseAddress.from_string(target_url)
+    target_url_obj = URL(target_url)
+    if_exists = target_url_obj.query_params.pop("if-exists", "fail")
+    cratedb_address = DatabaseAddress.from_string(str(target_url_obj))
     cratedb_url, cratedb_table = cratedb_address.decode()
-    if_exists = URL(target_url).query_params.get("if-exists") or "fail"
     if cratedb_table.table is None:
         raise ValueError("Table name is missing. Please adjust CrateDB database URL.")
     logger.info("Target address: %s", cratedb_address)

cratedb_toolkit/io/util.py

@@ -203,6 +203,13 @@ def schema(self) -> t.Union[str, None]:
         """
         return self.uri.query_params.get("schema") or self.uri.path.lstrip("/")
 
+    def with_table_address(self, table_address: "TableAddress") -> "DatabaseAddress":
+        cp = deepcopy(self)
+        cp.uri.path = f"/{table_address.schema}/{table_address.table}"
+        if table_address.if_exists:
+            cp.uri.query_params["if-exists"] = table_address.if_exists
+        return cp
+
 
 @dataclasses.dataclass
 class TableAddress:
@@ -212,6 +219,7 @@ class TableAddress:
 
     schema: t.Optional[str] = None
     table: t.Optional[str] = None
+    if_exists: t.Optional[str] = None
 
     def __bool__(self):
         return bool(self.table)

cratedb_toolkit/model.py

@@ -142,6 +142,7 @@ def get_connection_url(self, dialect: str = "crate", host: Optional[str] = None)
         """
         # TODO: When using `db_name=self.CRATEDB_DB`:
         #       Connection.__init__() got an unexpected keyword argument 'database'
+        #       Connection.__init__() got an unexpected keyword argument 'table'
         return super()._create_connection_url(
             dialect=dialect,
             username=self.CRATEDB_USER,

cratedb_toolkit/testing/testcontainers/cratedb.py

@@ -460,8 +460,9 @@ def decode_database_table(url: str) -> t.Tuple[t.Union[str, None], t.Union[str,
             if "too many values to unpack" not in str(ex) and "not enough values to unpack" not in str(ex):
                 raise
 
+        table = url_.query_params.get("table")
+        if not database:
             database = url_.query_params.get("database")
-            table = url_.query_params.get("table")
             if url_.scheme == "crate" and not database:
                 database = url_.query_params.get("schema")
             if database is None and table is None:

cratedb_toolkit/util/database.py

@@ -202,6 +202,7 @@ optional-dependencies.io-opentable = [
   "cratedb-toolkit[deltalake,iceberg]",
 ]
 optional-dependencies.io-recipe = [
+  "macropipe[geo]==0.0.0",
   "tikray>=0.2,<0.4",
 ]
 optional-dependencies.kinesis = [
@@ -292,6 +293,7 @@ line-length = 120
 line-length = 120
 extend-exclude = [
   "amqp-to-mqtt.py",
+  "examples",
   "workbench.py",
 ]
 lint.select = [