"""Synchronous and asynchronous HTTP clients for the CleanCSV API.
This module defines :class:`CleanCSVClient` and :class:`AsyncCleanCSVClient`, which
call the ``/api/{version}/clean`` and ``/api/{version}/suggest`` endpoints.
**File input type** ``FileInput``: a :class:`pathlib.Path` to a CSV file,
:obj:`bytes` / :obj:`bytearray`, or a binary file-like object whose
:meth:`~io.BufferedIOBase.read` returns :obj:`bytes`.
**API version** ``ApiVersion``: ``\"v1\"`` (production API) or ``\"demo\"`` (demo routes).
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import BinaryIO, Literal
from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
import httpx
from cleancsv.errors import CleanCSVError, RateLimitError
from cleancsv.types import CleanConfig, CleanResult, SuggestResponse
FileInput = Path | bytes | bytearray | BinaryIO
ApiVersion = Literal["v1", "demo"]
def _read_file_body(file: FileInput) -> bytes:
"""Read CSV payload as bytes from supported file-like inputs.
Args:
file: Path, raw bytes, or binary stream.
Returns:
File contents as bytes.
Raises:
TypeError: If a binary stream returns non-bytes from ``read()``.
"""
if isinstance(file, (bytes, bytearray)):
return bytes(file)
if isinstance(file, Path):
return file.read_bytes()
data = file.read()
if not isinstance(data, bytes):
msg = "BinaryIO must read bytes"
raise TypeError(msg)
return data
def _merge_query(url: str, params: dict[str, str]) -> str:
"""Merge query string parameters into a URL, preserving existing keys unless overridden.
Args:
url: Full URL that may already contain a query string.
params: Keys and values to merge into the query component.
Returns:
URL with merged query string.
"""
if not params:
return url
parsed = urlparse(url)
merged = dict(parse_qsl(parsed.query, keep_blank_values=True))
merged.update(params)
new_query = urlencode(list(merged.items()))
return urlunparse(parsed._replace(query=new_query))
def _error_from_response(resp: httpx.Response) -> CleanCSVError:
"""Build a :class:`~cleancsv.errors.CleanCSVError` from an error HTTP response.
Args:
resp: httpx response with a non-success status.
Returns:
Exception carrying message, optional JSON ``error`` field, status code, and body text.
"""
text = resp.text
try:
data = resp.json()
if isinstance(data, dict) and "error" in data:
return CleanCSVError(str(data["error"]), status_code=resp.status_code, body=text)
except json.JSONDecodeError:
pass
return CleanCSVError(text or f"HTTP {resp.status_code}", status_code=resp.status_code, body=text)
def _rate_limit_from_response(resp: httpx.Response) -> RateLimitError:
"""Build a :class:`~cleancsv.errors.RateLimitError` from an HTTP 429 response.
Parses ``X-RateLimit-Remaining``, ``Retry-After``, and optional JSON ``reset`` /
``remaining`` fields when present.
Args:
resp: httpx response with status 429.
Returns:
Rate limit exception with retry metadata when available.
"""
base = _error_from_response(resp)
remaining: int | None = None
reset_ms: int | None = None
retry_s: int | None = None
h = resp.headers
if "X-RateLimit-Remaining" in h:
try:
remaining = int(h["X-RateLimit-Remaining"])
except ValueError:
pass
if "Retry-After" in h:
try:
retry_s = int(h["Retry-After"])
except ValueError:
pass
try:
body = resp.json()
if isinstance(body, dict):
if "reset" in body:
reset_ms = int(body["reset"])
if remaining is None and "remaining" in body:
remaining = int(body["remaining"])
except json.JSONDecodeError:
pass
return RateLimitError(
base.args[0] if base.args else "Rate limit exceeded",
remaining=remaining,
reset_ms=reset_ms,
retry_after_s=retry_s,
body=base.body,
)
def _headers_to_map(h: httpx.Headers) -> dict[str, str]:
"""Flatten httpx headers (including duplicates) into a mapping.
Args:
h: Response or request headers.
Returns:
All header names and values as a dict (last value wins if duplicated).
"""
return {k: v for k, v in h.multi_items()}
[docs]
class CleanCSVClient:
"""Synchronous HTTP client for CleanCSV ``clean`` and ``suggest`` endpoints.
Pass the deployment **origin** only (no path): paths such as
``/api/v1/clean`` and ``/api/v1/suggest`` are appended by each method.
Use :data:`cleancsv.DEFAULT_BASE_URL` for the public app.
Use as a context manager to ensure the underlying connection is closed::
with CleanCSVClient(base_url) as client:
result = client.clean(path_to_csv)
Raises:
cleancsv.errors.CleanCSVError: On HTTP error responses from the API.
cleancsv.errors.RateLimitError: On HTTP 429 when rate limits apply.
"""
def __init__(
self,
base_url: str,
*,
timeout: float = 120.0,
client: httpx.Client | None = None,
) -> None:
"""Create a synchronous client.
Args:
base_url: Scheme and host (and optional port) of the CleanCSV deployment,
e.g. ``\"https://www.clean-csv.app\"``. Trailing slashes are stripped.
timeout: Default request timeout in seconds for the internal
:class:`httpx.Client` when ``client`` is not provided.
client: Optional pre-configured :class:`httpx.Client`. If omitted, a client
is created with ``timeout``.
"""
self._base = base_url.rstrip("/")
self._timeout = timeout
self._own_client = client is None
self._client = client or httpx.Client(timeout=timeout)
[docs]
def close(self) -> None:
"""Close the underlying HTTP client if this instance owns it."""
if self._own_client:
self._client.close()
[docs]
def __enter__(self) -> CleanCSVClient:
"""Enter context: return ``self``."""
return self
[docs]
def __exit__(self, *args: object) -> None:
"""Exit context: call :meth:`close`."""
self.close()
def _url(self, path: str) -> str:
return f"{self._base}{path}"
[docs]
def clean(
self,
file: FileInput,
*,
config: CleanConfig | str | None = None,
filename: str = "upload.csv",
version: ApiVersion = "v1",
query: dict[str, str] | None = None,
) -> CleanResult:
"""Upload a CSV via multipart POST to ``/api/{version}/clean``.
Sends the file as form field ``file`` and optional ``config`` as a JSON string
when ``config`` is set. URL query parameters such as ``format``, ``dedup_by``,
or ``strategy`` can be supplied via ``query`` (merged with any existing query).
Args:
file: CSV payload: path, bytes, or binary stream.
config: Cleaning options as :class:`~cleancsv.types.CleanConfig`, a JSON
string, or ``None`` to omit the config field.
filename: Filename sent in the multipart part (default ``upload.csv``).
version: API segment: ``\"v1\"`` or ``\"demo\"``.
query: Optional query parameters merged into the request URL.
Returns:
:class:`~cleancsv.types.CleanResult` with status, headers, raw body bytes,
and ``Content-Type``.
Raises:
cleancsv.errors.CleanCSVError: On non-success HTTP status (except 429).
cleancsv.errors.RateLimitError: On HTTP 429.
TypeError: If ``file`` cannot be read as bytes.
"""
path = f"/api/{version}/clean"
url = self._url(path)
if query:
url = _merge_query(url, query)
body = _read_file_body(file)
config_str: str | None
if config is None:
config_str = None
elif isinstance(config, str):
config_str = config
else:
config_str = config.to_json()
files = {"file": (filename, body, "text/csv")}
data = {"config": config_str} if config_str is not None else None
resp = self._client.post(url, files=files, data=data)
return _handle_clean_response_shared(resp)
[docs]
def clean_raw_body(
self,
body: bytes,
*,
content_type: str = "text/csv",
version: ApiVersion = "v1",
query: dict[str, str] | None = None,
) -> CleanResult:
"""POST raw CSV bytes to ``/api/{version}/clean`` without multipart encoding.
Use when the server accepts a raw request body. Query parameters can still
be passed via ``query``. This path does not send a multipart ``config`` field;
configure via URL query if the API supports it.
Args:
body: Raw CSV file bytes.
content_type: ``Content-Type`` header value (default ``text/csv``).
version: ``\"v1\"`` or ``\"demo\"``.
query: Optional query parameters merged into the request URL.
Returns:
:class:`~cleancsv.types.CleanResult` with response metadata and body.
Raises:
cleancsv.errors.CleanCSVError: On non-success HTTP status (except 429).
cleancsv.errors.RateLimitError: On HTTP 429.
"""
path = f"/api/{version}/clean"
url = self._url(path)
if query:
url = _merge_query(url, query)
resp = self._client.post(url, content=body, headers={"Content-Type": content_type})
return _handle_clean_response_shared(resp)
[docs]
def suggest(
self,
file: FileInput,
*,
filename: str = "upload.csv",
version: ApiVersion = "v1",
) -> SuggestResponse:
"""Upload a CSV via multipart POST to ``/api/{version}/suggest``.
Returns parsed JSON describing suggested deduplication rules and column stats.
Args:
file: CSV payload: path, bytes, or binary stream.
filename: Filename for the multipart part (default ``upload.csv``).
version: ``\"v1\"`` or ``\"demo\"``.
Returns:
:class:`~cleancsv.types.SuggestResponse` parsed from the JSON body.
Raises:
cleancsv.errors.CleanCSVError: On non-success HTTP status, invalid JSON,
or a non-object JSON body.
cleancsv.errors.RateLimitError: On HTTP 429.
TypeError: If ``file`` cannot be read as bytes.
"""
path = f"/api/{version}/suggest"
url = self._url(path)
body = _read_file_body(file)
files = {"file": (filename, body, "text/csv")}
resp = self._client.post(url, files=files)
return _handle_suggest_response_shared(resp)
[docs]
def suggest_raw_body(self, body: bytes, *, content_type: str = "text/csv", version: ApiVersion = "v1") -> SuggestResponse:
"""POST raw CSV bytes to ``/api/{version}/suggest`` without multipart encoding.
Args:
body: Raw CSV file bytes.
content_type: ``Content-Type`` header (default ``text/csv``).
version: ``\"v1\"`` or ``\"demo\"``.
Returns:
:class:`~cleancsv.types.SuggestResponse` parsed from the JSON body.
Raises:
cleancsv.errors.CleanCSVError: On HTTP errors or malformed JSON response.
cleancsv.errors.RateLimitError: On HTTP 429.
"""
url = self._url(f"/api/{version}/suggest")
resp = self._client.post(url, content=body, headers={"Content": content_type})
return _handle_suggest_response_shared(resp)
[docs]
class AsyncCleanCSVClient:
"""Asynchronous HTTP client for CleanCSV; mirrors :class:`CleanCSVClient` with async I/O.
Use ``async with`` to close the underlying client when it was created internally::
async with AsyncCleanCSVClient(base_url) as client:
result = await client.clean(path_to_csv)
Raises:
cleancsv.errors.CleanCSVError: On HTTP error responses from the API.
cleancsv.errors.RateLimitError: On HTTP 429 when rate limits apply.
"""
def __init__(
self,
base_url: str,
*,
timeout: float = 120.0,
client: httpx.AsyncClient | None = None,
) -> None:
"""Create an asynchronous client.
Args:
base_url: Deployment origin URL; trailing slashes are stripped.
timeout: Default timeout in seconds for a created :class:`httpx.AsyncClient`.
client: Optional existing async client; if omitted, one is created.
"""
self._base = base_url.rstrip("/")
self._timeout = timeout
self._own_client = client is None
self._client = client or httpx.AsyncClient(timeout=timeout)
[docs]
async def aclose(self) -> None:
"""Close the underlying async HTTP client if this instance owns it."""
if self._own_client:
await self._client.aclose()
[docs]
async def __aenter__(self) -> AsyncCleanCSVClient:
"""Enter async context: return ``self``."""
return self
[docs]
async def __aexit__(self, *args: object) -> None:
"""Exit async context: await :meth:`aclose`."""
await self.aclose()
def _url(self, path: str) -> str:
return f"{self._base}{path}"
[docs]
async def clean(
self,
file: FileInput,
*,
config: CleanConfig | str | None = None,
filename: str = "upload.csv",
version: ApiVersion = "v1",
query: dict[str, str] | None = None,
) -> CleanResult:
"""Async version of :meth:`CleanCSVClient.clean`.
Args:
file: CSV payload: path, bytes, or binary stream.
config: :class:`~cleancsv.types.CleanConfig`, JSON string, or ``None``.
filename: Multipart filename (default ``upload.csv``).
version: ``\"v1\"`` or ``\"demo\"``.
query: URL query parameters to merge.
Returns:
:class:`~cleancsv.types.CleanResult` with raw response body and metadata.
Raises:
cleancsv.errors.CleanCSVError: On HTTP errors (except 429).
cleancsv.errors.RateLimitError: On HTTP 429.
TypeError: If ``file`` cannot be read as bytes.
"""
path = f"/api/{version}/clean"
url = self._url(path)
if query:
url = _merge_query(url, query)
body = _read_file_body(file)
config_str: str | None
if config is None:
config_str = None
elif isinstance(config, str):
config_str = config
else:
config_str = config.to_json()
files = {"file": (filename, body, "text/csv")}
data = {"config": config_str} if config_str is not None else None
resp = await self._client.post(url, files=files, data=data)
return _handle_clean_response_shared(resp)
[docs]
async def clean_raw_body(
self,
body: bytes,
*,
content_type: str = "text/csv",
version: ApiVersion = "v1",
query: dict[str, str] | None = None,
) -> CleanResult:
"""Async version of :meth:`CleanCSVClient.clean_raw_body`.
Args:
body: Raw CSV bytes.
content_type: Request ``Content-Type``.
version: ``\"v1\"`` or ``\"demo\"``.
query: URL query parameters to merge.
Returns:
:class:`~cleancsv.types.CleanResult`.
Raises:
cleancsv.errors.CleanCSVError: On HTTP errors (except 429).
cleancsv.errors.RateLimitError: On HTTP 429.
"""
path = f"/api/{version}/clean"
url = self._url(path)
if query:
url = _merge_query(url, query)
resp = await self._client.post(url, content=body, headers={"Content-Type": content_type})
return _handle_clean_response_shared(resp)
[docs]
async def suggest(
self,
file: FileInput,
*,
filename: str = "upload.csv",
version: ApiVersion = "v1",
) -> SuggestResponse:
"""Async version of :meth:`CleanCSVClient.suggest`.
Args:
file: CSV payload: path, bytes, or binary stream.
filename: Multipart filename (default ``upload.csv``).
version: ``\"v1\"`` or ``\"demo\"``.
Returns:
:class:`~cleancsv.types.SuggestResponse`.
Raises:
cleancsv.errors.CleanCSVError: On HTTP errors or invalid JSON.
cleancsv.errors.RateLimitError: On HTTP 429.
TypeError: If ``file`` cannot be read as bytes.
"""
url = self._url(f"/api/{version}/suggest")
body = _read_file_body(file)
files = {"file": (filename, body, "text/csv")}
resp = await self._client.post(url, files=files)
return _handle_suggest_response_shared(resp)
[docs]
async def suggest_raw_body(
self, body: bytes, *, content_type: str = "text/csv", version: ApiVersion = "v1"
) -> SuggestResponse:
"""Async version of :meth:`CleanCSVClient.suggest_raw_body`.
Args:
body: Raw CSV bytes.
content_type: Request ``Content-Type``.
version: ``\"v1\"`` or ``\"demo\"``.
Returns:
:class:`~cleancsv.types.SuggestResponse`.
Raises:
cleancsv.errors.CleanCSVError: On HTTP errors or malformed JSON.
cleancsv.errors.RateLimitError: On HTTP 429.
"""
url = self._url(f"/api/{version}/suggest")
resp = await self._client.post(url, content=body, headers={"Content-Type": content_type})
return _handle_suggest_response_shared(resp)
def _handle_clean_response_shared(resp: httpx.Response) -> CleanResult:
"""Normalize a successful ``clean`` HTTP response into :class:`~cleancsv.types.CleanResult`.
Args:
resp: httpx response from a clean endpoint.
Returns:
Result object with ``status_code``, ``content_type``, ``body``, and ``headers``.
Raises:
cleancsv.errors.RateLimitError: If status is 429.
cleancsv.errors.CleanCSVError: If the response is another HTTP error.
"""
if resp.status_code == 429:
raise _rate_limit_from_response(resp)
if resp.is_error:
raise _error_from_response(resp)
ct = resp.headers.get("content-type", "").split(";")[0].strip()
return CleanResult(
status_code=resp.status_code,
content_type=ct or "application/octet-stream",
body=resp.content,
headers=dict(resp.headers),
)
def _handle_suggest_response_shared(resp: httpx.Response) -> SuggestResponse:
"""Parse a successful ``suggest`` JSON response into :class:`~cleancsv.types.SuggestResponse`.
Args:
resp: httpx response from a suggest endpoint.
Returns:
Parsed structured suggest payload.
Raises:
cleancsv.errors.RateLimitError: If status is 429.
cleancsv.errors.CleanCSVError: If HTTP error or JSON is not a JSON object.
"""
if resp.status_code == 429:
raise _rate_limit_from_response(resp)
if resp.is_error:
raise _error_from_response(resp)
data = resp.json()
if not isinstance(data, dict):
msg = "Suggest response must be a JSON object"
raise CleanCSVError(msg, status_code=resp.status_code, body=resp.text)
return SuggestResponse.from_api(data)