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/io/file/__init__.py

@@ -0,0 +1,124 @@
+"""
+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
+    batch_size: Optional[int] = DEFAULT_BATCH_SIZE
+    separator: Optional[str] = DEFAULT_SEPARATOR
+    quote_char: Optional[str] = DEFAULT_QUOTE_CHAR
+    pipeline: Optional[List[str]] = dataclasses.field(default_factory=list)
+
+    @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,
+            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),
+            pipeline=url_obj.query_params.getlist("pipe"),
+        )
+
+    @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 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

@@ -212,6 +212,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

@@ -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 = [

pyproject.toml

@@ -0,0 +1,14 @@
+CREATE TABLE "{schema}".climate_data
+(
+    "timestamp"    TIMESTAMP WITHOUT TIME ZONE,
+    "geo_location" GEO_POINT,
+    "data"         OBJECT(DYNAMIC) AS (
+        "temperature"  DOUBLE PRECISION,
+        "u10"          DOUBLE PRECISION,
+        "v10"          DOUBLE PRECISION,
+        "pressure"     DOUBLE PRECISION,
+        "latitude"     DOUBLE PRECISION,
+        "longitude"    DOUBLE PRECISION,
+        "humidity"     DOUBLE PRECISION
+    )
+);