"""Configuration and response datatypes for the CleanCSV API.
These types mirror the JSON shape accepted by the clean endpoint and returned by
clean/suggest responses. Use :class:`CleanConfig` with :class:`cleancsv.client.CleanCSVClient`
or build configs from dicts via :func:`config_from_mapping`.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Literal, Mapping
Format = Literal["json", "csv"]
DedupStrategy = Literal["first", "last", "most_complete", "merge"]
UnicodeNorm = Literal["NFC", "NFD", "NFKC", "NFKD"]
LineEndings = Literal["unix", "windows", "auto"]
CastType = Literal["string", "int", "float", "boolean", "auto"]
FillStrategy = str # leave, drop_row, ... or fill_default:...
[docs]
@dataclass
class MojibakeMapping:
"""Single replacement rule for fixing mojibake in :class:`EncodingConfig`.
Attributes:
from_: Source byte/character sequence to replace (JSON key ``from``).
to: Replacement string sent to the API as ``to``.
"""
from_: str
to: str
[docs]
def to_api(self) -> dict[str, str]:
"""Serialize to the API JSON object shape (``from`` / ``to`` keys).
Returns:
Dict with keys ``\"from\"`` and ``\"to\"`` suitable for ``encoding.custom_mojibake_mappings``.
"""
return {"from": self.from_, "to": self.to}
[docs]
@dataclass
class DedupConfig:
"""Deduplication options under ``dedup`` in :class:`CleanConfig`.
Attributes:
columns: Column names to use for duplicate detection; ``None`` means API default.
strategy: How to pick the surviving row among duplicates.
fuzzy: Whether to use fuzzy matching.
threshold: Fuzzy similarity threshold (e.g. ``0.85``).
normalize: Whether to normalize values before comparing.
preview: If true, API may return preview rows without full output.
nulls_equal: Whether null/empty values match each other for dedup.
"""
columns: list[str] | None = None
strategy: DedupStrategy = "first"
fuzzy: bool = False
threshold: float = 0.85
normalize: bool = False
preview: bool = False
nulls_equal: bool = True
[docs]
def to_api(self) -> dict[str, Any]:
"""Convert to a JSON-serializable dict for the ``dedup`` section of clean config.
Returns:
Mapping ready for inclusion in :meth:`CleanConfig.to_api_dict`.
"""
d: dict[str, Any] = {
"strategy": self.strategy,
"fuzzy": self.fuzzy,
"threshold": self.threshold,
"normalize": self.normalize,
"preview": self.preview,
"nulls_equal": self.nulls_equal,
}
if self.columns is not None:
d["columns"] = self.columns
return d
[docs]
@dataclass
class EncodingConfig:
"""Encoding and text-normalization options under ``encoding`` in :class:`CleanConfig`.
Attributes:
encoding: Source encoding name or ``\"auto\"``.
fallback: Optional list of encodings to try when detection is ambiguous.
fix_mojibake: Enable mojibake repair heuristics.
custom_mojibake_mappings: Optional list of :class:`MojibakeMapping` rules.
normalize_unicode: If ``True`` or a Unicode form name, apply normalization.
strip_control_chars: Strip control characters from cell text.
strip_invisible: Strip invisible Unicode characters.
fix_bom: Handle byte-order mark.
line_endings: Normalized line ending style: ``unix``, ``windows``, or ``auto``.
"""
encoding: str = "auto"
fallback: list[str] | None = None
fix_mojibake: bool = True
custom_mojibake_mappings: list[MojibakeMapping] | None = None
normalize_unicode: bool | UnicodeNorm = True
strip_control_chars: bool = True
strip_invisible: bool = True
fix_bom: bool = True
line_endings: LineEndings = "unix"
[docs]
def to_api(self) -> dict[str, Any]:
"""Convert to JSON-serializable dict for the ``encoding`` section.
Returns:
Mapping for :meth:`CleanConfig.to_api_dict`.
"""
d: dict[str, Any] = {
"encoding": self.encoding,
"fix_mojibake": self.fix_mojibake,
"normalize_unicode": self.normalize_unicode,
"strip_control_chars": self.strip_control_chars,
"strip_invisible": self.strip_invisible,
"fix_bom": self.fix_bom,
"line_endings": self.line_endings,
}
if self.fallback is not None:
d["fallback"] = self.fallback
if self.custom_mojibake_mappings:
d["custom_mojibake_mappings"] = [m.to_api() for m in self.custom_mojibake_mappings]
return d
[docs]
@dataclass
class NullColumnConfig:
"""Per-column null handling under ``null_handling.per_column``.
Attributes:
treat_empty_as_null: Whether empty strings count as null for this column.
null_values: Extra string values to treat as null.
fill: Fill strategy (e.g. ``leave``, ``drop_row``, or API-specific tokens).
cast: Target type for casting: ``string``, ``int``, ``float``, ``boolean``, ``auto``.
"""
treat_empty_as_null: bool | None = None
null_values: list[str] | None = None
fill: FillStrategy | None = None
cast: CastType | None = None
[docs]
def to_api(self) -> dict[str, Any]:
"""Emit only keys that are set (omits ``None`` fields).
Returns:
Dict fragment for one column in ``per_column``.
"""
out: dict[str, Any] = {}
if self.treat_empty_as_null is not None:
out["treat_empty_as_null"] = self.treat_empty_as_null
if self.null_values is not None:
out["null_values"] = self.null_values
if self.fill is not None:
out["fill"] = self.fill
if self.cast is not None:
out["cast"] = self.cast
return out
[docs]
@dataclass
class NullHandlingConfig:
"""Global and per-column null handling under ``null_handling`` in :class:`CleanConfig`.
Attributes:
null_values: Strings treated as null globally.
treat_empty_as_null: Whether empty cells are null by default.
type_inference: Whether to infer column types.
fill: Default fill strategy for nulls.
per_column: Optional map of column name to :class:`NullColumnConfig`.
"""
null_values: list[str] | None = None
treat_empty_as_null: bool = True
type_inference: bool = True
fill: FillStrategy = "leave"
per_column: dict[str, NullColumnConfig] | None = None
[docs]
def to_api(self) -> dict[str, Any]:
"""Convert to JSON-serializable dict for the ``null_handling`` section.
Returns:
Mapping for :meth:`CleanConfig.to_api_dict`.
"""
d: dict[str, Any] = {
"treat_empty_as_null": self.treat_empty_as_null,
"type_inference": self.type_inference,
"fill": self.fill,
}
if self.null_values is not None:
d["null_values"] = self.null_values
if self.per_column:
d["per_column"] = {k: v.to_api() for k, v in self.per_column.items()}
return d
[docs]
@dataclass
class CleanConfig:
"""Cleaning options sent as the multipart form field ``config`` (JSON string).
Corresponds to the CleanCSV upload/config schema: output ``format``, optional
``dedup``, ``encoding``, ``null_handling``, email filters, etc.
Attributes:
format: Response format for the clean endpoint: ``json`` or ``csv``.
dedup: Optional :class:`DedupConfig`.
encoding: Optional :class:`EncodingConfig`.
null_handling: Optional :class:`NullHandlingConfig`.
null_representation: How nulls appear in output (API-specific string).
filter_invalid_emails: Drop or flag rows with invalid emails when enabled.
email_columns: Column names to validate as email when filtering is on.
"""
format: Format = "json"
dedup: DedupConfig | None = None
encoding: EncodingConfig | None = None
null_handling: NullHandlingConfig | None = None
null_representation: str | None = None
filter_invalid_emails: bool = False
email_columns: list[str] | None = None
[docs]
def to_api_dict(self) -> dict[str, Any]:
"""Build the full config object as a plain dict for JSON encoding.
Returns:
Nested dict matching the API's expected ``config`` JSON structure.
"""
cfg: dict[str, Any] = {"format": self.format}
if self.dedup is not None:
cfg["dedup"] = self.dedup.to_api()
if self.encoding is not None:
cfg["encoding"] = self.encoding.to_api()
if self.null_handling is not None:
cfg["null_handling"] = self.null_handling.to_api()
if self.null_representation is not None:
cfg["null_representation"] = self.null_representation
if self.filter_invalid_emails:
cfg["filter_invalid_emails"] = True
if self.email_columns is not None:
cfg["email_columns"] = self.email_columns
return cfg
[docs]
def to_json(self) -> str:
"""Serialize :meth:`to_api_dict` to a compact JSON string (no extra whitespace).
Returns:
JSON text suitable for the multipart ``config`` field.
"""
import json
return json.dumps(self.to_api_dict(), separators=(",", ":"))
[docs]
@classmethod
def from_suggested_rule(cls, rule: "SuggestedRule", *, format: Format = "json") -> CleanConfig:
"""Build a minimal clean config from one :class:`SuggestedRule` (e.g. after ``suggest``).
Args:
rule: A suggested rule from :class:`SuggestResponse`.
format: Desired clean output format (``json`` or ``csv``).
Returns:
:class:`CleanConfig` with ``dedup`` populated from the rule's columns and flags.
"""
return cls(
format=format,
dedup=DedupConfig(
columns=list(rule.dedup_by),
strategy="first",
fuzzy=rule.fuzzy,
threshold=rule.threshold if rule.threshold is not None else 0.85,
normalize=rule.normalize,
),
)
[docs]
@dataclass
class SuggestedRule:
"""One deduplication suggestion from the ``/suggest`` API.
Attributes:
dedup_by: Column names the API recommends for duplicate detection.
fuzzy: Whether fuzzy matching was suggested.
normalize: Whether normalization was suggested.
reason: Short explanation from the API.
threshold: Suggested fuzzy threshold, if any.
"""
dedup_by: list[str]
fuzzy: bool
normalize: bool
reason: str
threshold: float | None = None
[docs]
@classmethod
def from_api(cls, raw: Mapping[str, Any]) -> SuggestedRule:
"""Parse a single rule object from API JSON (camelCase keys).
Args:
raw: One element from the ``suggested_rules`` array.
Returns:
:class:`SuggestedRule` instance.
"""
return cls(
dedup_by=list(raw["dedupBy"]),
fuzzy=bool(raw["fuzzy"]),
normalize=bool(raw["normalize"]),
reason=str(raw["reason"]),
threshold=float(raw["threshold"]) if raw.get("threshold") is not None else None,
)
[docs]
@dataclass
class SuggestResponse:
"""Structured JSON body from ``POST .../suggest``.
Attributes:
row_count: Number of data rows analyzed.
column_count: Number of columns.
columns: Column names in order.
suggested_rules: List of :class:`SuggestedRule` entries.
null_profiles: Optional per-column null statistics from the API.
warnings: Optional list of warning objects from the API.
"""
row_count: int
column_count: int
columns: list[str]
suggested_rules: list[SuggestedRule]
null_profiles: list[dict[str, Any]] | None = None
warnings: list[dict[str, Any]] | None = None
[docs]
@classmethod
def from_api(cls, raw: Mapping[str, Any]) -> SuggestResponse:
"""Parse the top-level suggest response JSON.
Args:
raw: Parsed JSON object from the suggest endpoint.
Returns:
:class:`SuggestResponse` with nested rules converted to dataclasses.
"""
rules = [SuggestedRule.from_api(r) for r in raw.get("suggested_rules", [])]
return cls(
row_count=int(raw["row_count"]),
column_count=int(raw["column_count"]),
columns=list(raw["columns"]),
suggested_rules=rules,
null_profiles=list(raw["null_profiles"]) if raw.get("null_profiles") else None,
warnings=list(raw["warnings"]) if raw.get("warnings") else None,
)
[docs]
@dataclass
class CleanResult:
"""Raw HTTP result from ``POST .../clean`` before parsing structured JSON/CSV.
Attributes:
status_code: HTTP status (typically 200 on success).
content_type: Normalized primary ``Content-Type`` (without parameters).
body: Raw response bytes (JSON or CSV depending on config).
headers: Response headers as a flat mapping.
"""
status_code: int
content_type: str
body: bytes
headers: Mapping[str, str] = field(default_factory=dict)
@property
def text(self) -> str:
"""Decode :attr:`body` as UTF-8 with replacement for invalid sequences.
Returns:
Unicode text of the response body.
"""
return self.body.decode("utf-8", errors="replace")
[docs]
def parse_json(self) -> CleanResponse:
"""Parse the body as JSON into :class:`CleanResponse` when format is JSON.
Returns:
Structured clean response with ``meta``, ``data``, and optional preview fields.
Raises:
ValueError: If ``Content-Type`` does not indicate JSON or JSON is invalid.
"""
import json
if "json" not in self.content_type.lower():
msg = f"Expected JSON response, got Content-Type={self.content_type!r}"
raise ValueError(msg)
data = json.loads(self.text)
return CleanResponse.from_api(data)
[docs]
@dataclass
class CleanResponse:
"""Parsed JSON body from a successful JSON-format clean operation.
Attributes:
meta: Opaque metadata dict from the API (e.g. stats, timing).
data: Cleaned rows as list of column-name to value dicts.
duplicate_groups: Optional duplicate grouping information.
preview_source: Optional preview of source rows.
preview_cleaned: Optional preview of cleaned rows.
preview_row_in_output: Optional flags per preview row.
preview_row_removal_reason: Optional reasons for excluded preview rows.
"""
meta: dict[str, Any]
data: list[dict[str, Any]]
duplicate_groups: list[dict[str, Any]] | None = None
preview_source: list[dict[str, Any]] | None = None
preview_cleaned: list[dict[str, Any]] | None = None
preview_row_in_output: list[bool] | None = None
preview_row_removal_reason: list[str | None] | None = None
[docs]
@classmethod
def from_api(cls, raw: Mapping[str, Any]) -> CleanResponse:
"""Parse the top-level clean JSON response.
Args:
raw: Parsed JSON object from the clean endpoint (JSON format).
Returns:
:class:`CleanResponse` with optional preview fields set when present.
"""
return cls(
meta=dict(raw["meta"]),
data=list(raw["data"]),
duplicate_groups=list(raw["duplicate_groups"]) if raw.get("duplicate_groups") else None,
preview_source=list(raw["preview_source"]) if raw.get("preview_source") else None,
preview_cleaned=list(raw["preview_cleaned"]) if raw.get("preview_cleaned") else None,
preview_row_in_output=list(raw["preview_row_in_output"]) if raw.get("preview_row_in_output") else None,
preview_row_removal_reason=list(raw["preview_row_removal_reason"])
if raw.get("preview_row_removal_reason")
else None,
)
[docs]
def config_from_mapping(m: Mapping[str, Any]) -> CleanConfig:
"""Build :class:`CleanConfig` from a plain dict (e.g. loaded from JSON on disk).
Recognizes the same keys as the API: ``format``, ``dedup``, ``encoding``,
``null_handling``, ``null_representation``, ``filter_invalid_emails``, ``email_columns``.
Args:
m: Configuration mapping in API snake_case shape.
Returns:
Populated :class:`CleanConfig` with nested dataclasses where sections exist.
"""
dedup = m.get("dedup")
enc = m.get("encoding")
nh = m.get("null_handling")
d_cfg = None
if isinstance(dedup, Mapping):
d_cfg = DedupConfig(
columns=list(dedup["columns"]) if dedup.get("columns") is not None else None,
strategy=dedup.get("strategy", "first"), # type: ignore[arg-type]
fuzzy=bool(dedup.get("fuzzy", False)),
threshold=float(dedup.get("threshold", 0.85)),
normalize=bool(dedup.get("normalize", False)),
preview=bool(dedup.get("preview", False)),
nulls_equal=dedup.get("nulls_equal", True) is not False,
)
e_cfg = None
if isinstance(enc, Mapping):
mappings = enc.get("custom_mojibake_mappings")
mm = None
if isinstance(mappings, list):
mm = [MojibakeMapping(from_=x["from"], to=x["to"]) for x in mappings if isinstance(x, Mapping)]
e_cfg = EncodingConfig(
encoding=str(enc.get("encoding", "auto")),
fallback=list(enc["fallback"]) if isinstance(enc.get("fallback"), list) else None,
fix_mojibake=enc.get("fix_mojibake", True) is not False,
custom_mojibake_mappings=mm,
normalize_unicode=enc.get("normalize_unicode", True), # type: ignore[arg-type]
strip_control_chars=enc.get("strip_control_chars", True) is not False,
strip_invisible=enc.get("strip_invisible", True) is not False,
fix_bom=enc.get("fix_bom", True) is not False,
line_endings=enc.get("line_endings", "unix"), # type: ignore[arg-type]
)
n_cfg = None
if isinstance(nh, Mapping):
per = nh.get("per_column")
pc: dict[str, NullColumnConfig] | None = None
if isinstance(per, Mapping):
pc = {}
for col, v in per.items():
if isinstance(v, Mapping):
pc[str(col)] = NullColumnConfig(
treat_empty_as_null=v.get("treat_empty_as_null")
if isinstance(v.get("treat_empty_as_null"), bool)
else None,
null_values=list(v["null_values"]) if isinstance(v.get("null_values"), list) else None,
fill=str(v["fill"]) if v.get("fill") is not None else None,
cast=v.get("cast"), # type: ignore[arg-type]
)
n_cfg = NullHandlingConfig(
null_values=list(nh["null_values"]) if isinstance(nh.get("null_values"), list) else None,
treat_empty_as_null=nh.get("treat_empty_as_null", True) is not False,
type_inference=nh.get("type_inference", True) is not False,
fill=str(nh.get("fill", "leave")),
per_column=pc,
)
fmt: Format = "csv" if m.get("format") == "csv" else "json"
return CleanConfig(
format=fmt,
dedup=d_cfg,
encoding=e_cfg,
null_handling=n_cfg,
null_representation=str(m["null_representation"]) if m.get("null_representation") is not None else None,
filter_invalid_emails=bool(m.get("filter_invalid_emails", False)),
email_columns=list(m["email_columns"]) if isinstance(m.get("email_columns"), list) else None,
)