feat: python-csv node kind (contract-only CSV loader + validation header check) - #42
Conversation
Signed-off-by: Simone Carolini <simonecarolini.sc@gmail.com>
Signed-off-by: Simone Carolini <simonecarolini.sc@gmail.com>
…st minio and a live HTTP server Signed-off-by: Simone Carolini <simonecarolini.sc@gmail.com>
… extend/overflow tests Signed-off-by: Simone Carolini <simonecarolini.sc@gmail.com>
…rced Signed-off-by: Simone Carolini <simonecarolini.sc@gmail.com>
…sh, no closure Signed-off-by: Simone Carolini <simonecarolini.sc@gmail.com>
Signed-off-by: Simone Carolini <simonecarolini.sc@gmail.com>
…in build_from_columns Signed-off-by: Simone Carolini <simonecarolini.sc@gmail.com>
Signed-off-by: Simone Carolini <simonecarolini.sc@gmail.com>
…un warning; docs(boundary): kind field
Final fix wave from the Phase-A python-csv whole-branch review:
- FIX 1: HttpsCsvSourceReader.fetch_header_line no longer loops forever
re-appending an identical full body when a server ignores Range and
answers 200. It now checks the response status: a 200 (not 206) is
terminal on the first pass regardless of size or newline presence,
fixing a false MAX_HEADER_BYTES overflow for a no-newline body sized
between HEADER_PROBE_BYTES and MAX_HEADER_BYTES.
- FIX 6: validation/runner.py raises a clear ValueError naming the
csv_source when fetch_header_line returns an empty header line,
instead of an opaque StopIteration from csv.reader. Discovered while
writing the real-minio test that a 0-byte S3 object's first Range
probe raises botocore InvalidRange rather than returning an empty
body, so S3CsvSourceReader.fetch_header_line now also treats an
InvalidRange error on the first probe as an empty object.
- FIX 7: csv_loader.produce_csv now logs the same structured
csv_header_extra_columns warning the validation runner emits, giving
extra_columns: drop spec parity between the validation and run paths.
- FIX 2: collapse the "reads must be exactly {csv: <uri>}" error message
into one f-string; message text unchanged.
- FIX 3: add a test covering the existing exit-2 guard for a non-string
csv_source in the candidate spec.
- docs(boundary-contract): document the additive `kind` wire field
(default python-model; python-csv is a contract-only node whose reads
is exactly {csv: <uri>} and whose source_hash is sha256 of that uri).
Signed-off-by: Simone Carolini <simonecarolini.sc@gmail.com>
carolsimone
left a comment
carolsimone
left a comment
There was a problem hiding this comment.
The CSV producer can corrupt or reject valid values through premature type inference, and the HTTPS reader introduces security and resource-exhaustion paths. Several validation edge cases also bypass the intended error taxonomy or release checks.
| try: | ||
| with tempfile.TemporaryDirectory() as tmp: | ||
| dest = active_reader.fetch(uri, Path(tmp) / "source.csv") | ||
| table = pyarrow.csv.read_csv(dest) |
There was a problem hiding this comment.
[P1] Preserve lexical CSV values before conformance - When a declared TEXT/VARCHAR column contains numeric-looking values such as 00123, read_csv infers int64 first and conform() loads 123; valid DECIMAL values are similarly inferred as floats and then rejected as lossy casts. Parse declared columns as strings or with an explicit conversion strategy so conformance receives the original values.
There was a problem hiding this comment.
Confirmed: pyarrow's default inference read 00123 as int64 and NUMERIC values as float64 (rejected by conform()'s lossy-cast guard). Fixed in be30a0e — csv_loader.py now passes output_columns as read_csv's ConvertOptions.column_types, so every declared column is parsed directly as its target Arrow type (full mapping via the existing types.arrow_type/parse_sql_type, not just VARCHAR/TEXT — pyarrow's CSV reader handles decimal128/date32/timestamp fine as explicit convert targets). New tests cover the 00123-VARCHAR and NUMERIC(10,2) cases end to end.
| end = start + HEADER_PROBE_BYTES - 1 | ||
| req = urllib.request.Request( | ||
| uri.raw, headers={"Range": f"bytes={start}-{end}"}) | ||
| with urllib.request.urlopen(req) as resp: # noqa: S310 — scheme gated by parse_csv_uri |
There was a problem hiding this comment.
[P1] Revalidate HTTPS schemes after redirects - When an accepted HTTPS endpoint redirects to HTTP or FTP, urllib follows it automatically, so both the header probe and full fetch bypass the parse_csv_uri scheme restriction. This permits plaintext downgrade and access to HTTP-only internal targets; use a redirect handler that only permits HTTPS destinations.
There was a problem hiding this comment.
Confirmed: urllib followed a redirect regardless of scheme, so an https source could downgrade to http. Fixed in d1bf4a0 — installed a _HttpsOnlyRedirectHandler that raises unless the redirect target itself starts with https://, used by both urlopen calls via a shared opener. Tested end to end against a REACHABLE downgrade target (not just an unreachable host) so the test actually proves the pre-fix code followed the redirect, not merely that some exception occurred.
| req = urllib.request.Request( | ||
| uri.raw, headers={"Range": f"bytes={start}-{end}"}) | ||
| with urllib.request.urlopen(req) as resp: # noqa: S310 — scheme gated by parse_csv_uri | ||
| body = resp.read() |
There was a problem hiding this comment.
[P1] Bound header reads when servers ignore Range - When an HTTPS server ignores Range and returns 200, this unbounded read() buffers the entire CSV before the status is inspected. A multi-gigabyte source can therefore exhaust the validation process even though only its header is needed; stream only through the newline or MAX_HEADER_BYTES + 1.
There was a problem hiding this comment.
Confirmed: resp.read() ran unconditionally, before the status check, so a range-ignored (200) response buffered the whole object. Fixed in d1bf4a0 — the range-ignored path now reads via a bounded chunked helper that stops at the first newline or MAX_HEADER_BYTES, so a multi-gigabyte body is never buffered in full.
| raise | ||
| buf += body | ||
| if b"\n" in buf: | ||
| return buf.split(b"\n", 1)[0].rstrip(b"\r").decode("utf-8") |
There was a problem hiding this comment.
[P2] Enforce the header-size limit before returning - If the first newline arrives after MAX_HEADER_BYTES, this branch returns the oversized header before checking the buffer length. The same ordering exists in the HTTPS reader, so headers larger than the documented 1 MiB limit are accepted whenever they contain a newline; validate the pre-newline length before returning.
There was a problem hiding this comment.
Confirmed, and it applied to https.py too as you noted. Fixed in d1bf4a0 — extracted extract_header_line() (in csv_source.py, the dependency-free port module) which measures the resolved line itself (bytes up to the first newline) against MAX_HEADER_BYTES before returning, instead of checking only the intermediate accumulated-buffer length. Both s3.py and https.py now use it. New test: a header line that terminates in a newline only after exceeding the limit — previously returned as an oversized 'success', now raises.
| start += HEADER_PROBE_BYTES | ||
|
|
||
| def fetch(self, uri: CsvUri, dest: Path) -> Path: | ||
| with urllib.request.urlopen(uri.raw) as resp, open(dest, "wb") as f: # noqa: S310 |
There was a problem hiding this comment.
[P2] Add finite timeouts to HTTPS fetches - When an HTTPS source accepts a connection but stalls while sending headers or data, both urlopen calls use the effectively unbounded default timeout, allowing validation or a scheduled node run to hang indefinitely. Pass an explicit finite timeout to the header and full-object requests.
There was a problem hiding this comment.
Fixed in d1bf4a0 — added a module-level _TIMEOUT_SECONDS = 30 constant, passed to both urlopen calls (header probe and full fetch) via the shared opener.
| f"{csv_source}" | ||
| ) | ||
| declared = [c["name"] for c in spec["output_columns"]] | ||
| extras = check_header(next(csv.reader([header_line])), declared) |
There was a problem hiding this comment.
[P2] Normalize BOMs before validating CSV headers - For a UTF-8-BOM CSV, PyArrow strips the BOM and the runtime sees order_id, but the header readers decode with plain UTF-8 and csv.reader produces U+FEFForder_id. The release gate consequently rejects a source the runtime can load; decode the first header as utf-8-sig or strip the BOM before check_header.
There was a problem hiding this comment.
Confirmed. Fixed in d1bf4a0 — both readers now decode header bytes as utf-8-sig everywhere a header line is produced (in the shared extract_header_line() and the two no-newline-short-body return paths), so a BOM'd source's first column matches the declared name. New BOM test for both readers.
| raise ContractError(f"{label}: unknown key(s) {sorted(unknown)}") | ||
|
|
||
| kind = raw.get("kind", "python-model") | ||
| if kind not in KINDS: |
There was a problem hiding this comment.
[P2] Type-check kind before set membership - When malformed YAML supplies an unhashable kind such as kind: [python-csv], membership in the frozenset raises a bare TypeError before the intended ContractError. This escapes CLI error handling and produces a traceback instead of a validation failure; reject non-string kinds first.
There was a problem hiding this comment.
Confirmed: kind: [python-csv] raised a bare TypeError from the frozenset membership test on an unhashable value, before ContractError. Fixed in d4044ac — check isinstance(kind, str) first (short-circuits before the membership test). New parametrized test covers list/dict/int/bool/None.
| Raises ValueError for anything else (http:// included) so contract | ||
| validation fails at lint/parse time, never at run time. | ||
| """ | ||
| if uri.startswith("s3://"): |
There was a problem hiding this comment.
[P2] Reject non-string CSV URIs explicitly - For a contract such as reads: {csv: 123}, this call raises AttributeError from startswith() rather than ValueError or TypeError, so the loader wrapper does not convert it to ContractError and validation crashes. Check that the URI is a string before examining its prefix.
There was a problem hiding this comment.
Confirmed: reads: {csv: 123} raised AttributeError from .startswith(), uncaught by loader.py's except (ValueError, TypeError). Fixed in d1bf4a0 — parse_csv_uri now raises ValueError for a non-string uri up front. New parametrized test (including a bytes value, which raised a different TypeError than list/dict/int did on the old code).
| sys.exit(2) | ||
| config = raw_config or {} | ||
| csv_source = spec.get("csv_source", "") | ||
| if csv_source and not isinstance(csv_source, str): |
There was a problem hiding this comment.
[P2] Reject falsey non-string csv_source values - If a candidate spec contains csv_source: false, 0, [], or {}, the truthiness guard skips both the type error and the later header check. A malformed python-csv candidate can therefore pass promotion without validating its source; validate the field type independently of its truthiness.
There was a problem hiding this comment.
Confirmed: csv_source: 0/false/[]/{} skipped the truthiness-gated check entirely, so no type error and no header check. Fixed in d4044ac — gate on 'csv_source' in spec (presence) instead of truthiness. New parametrized test (0, false, [], {}) asserts exit 2 for all four.
| from continuo_python_runtime.csv_readers.https import HttpsCsvSourceReader | ||
| from continuo_python_runtime.csv_readers.s3 import S3CsvSourceReader | ||
|
|
||
| pytestmark = pytest.mark.integration |
There was a problem hiding this comment.
[P2] Keep Docker tests out of the default runtime suite - These tests are marked integration, but the documented and CI runtime command only excludes image, so the new MinIO-backed reader and runner tests remain selected and invoke Docker. On the supported local setup without Docker, the standard pre-PR command now errors despite CONTRIBUTING.md lines 41-42 saying Docker is only needed for adapter integration tests; exclude integration tests from that suite or move them to a dedicated job.
There was a problem hiding this comment.
Confirmed — CONTRIBUTING.md's own documented pre-PR command had the same gap. Went with option (a): fixed in 0a1d215. The runtime step's marker is now -m "not image and not integration", and a new 'Tests (runtime, integration)' step runs tests/test_csv_readers_integration.py and tests/test_validation_runner.py under -m integration explicitly on the same runner (docker is available; minio self-provisions via the fixture's own docker run). CONTRIBUTING.md updated to match. Verified nothing is silently dropped: 463 (main) + 30 (integration) + 3 (image, unchanged in images.yml) = 496 total collected.
pyarrow's default type inference in read_csv was the first thing to interpret a value, ahead of conform(): a VARCHAR column holding 00123 inferred as int64 and conform() wrote back "123", and a NUMERIC column holding a valid decimal like 10.50 inferred as float64, which conform()'s own lossy-cast guard then rejected outright. Pass output_columns as read_csv's ConvertOptions.column_types so every declared column is parsed once, directly as the type it is declared to be. Signed-off-by: Simone Carolini <simonecarolini.sc@gmail.com>
…g, BOM Four independent hardening fixes to the S3/HTTPS csv source readers: - https.py: an https:// source that redirected to a non-https target was followed silently (urllib doesn't care about scheme on redirect), downgrading both the header probe and the full fetch to plaintext. Install a redirect handler that refuses any target not itself https://. - https.py: when a server ignores our Range header and answers 200, the response is the whole object -- reading it with an unbounded resp.read() before even checking the status buffered a multi-gigabyte body in full just to inspect its header line. Bound that read to stop at the first newline or MAX_HEADER_BYTES. - s3.py/https.py: a newline that only arrived after the accumulated probe buffer had already grown past MAX_HEADER_BYTES was returned as a successful (oversized) header line, because the newline check ran before the length check. Extracted the shared extract_header_line() (in csv_source.py, the dependency-free port module) which measures the *resolved line itself* against the limit, and both readers now use it. - s3.py/https.py: a UTF-8 byte-order mark on the CSV's first byte survived a plain "utf-8" decode as a literal U+FEFF prepended to the first column name, so a declared column matching the visually-identical name failed check_header. Decode as utf-8-sig everywhere a header line is produced. Also adds an explicit finite timeout (30s) to both HTTPS urlopen calls: an https source that accepts the connection but stalls on headers or body would otherwise hang a validation or node run indefinitely. Tests: test_csv_readers_integration.py gains oversized-line-with-late- newline and BOM cases for both readers, plus a redirect-to-a-reachable- downgrade-target test proving the pre-fix code actually followed it (rather than merely erroring on an unreachable host). test_csv_readers_ https.py is a new unit tier covering the redirect handler and timeout wiring without a live server. Signed-off-by: Simone Carolini <simonecarolini.sc@gmail.com>
…rash
- contract/loader.py: kind: [python-csv] (or any unhashable kind) raised a
bare TypeError from `kind not in KINDS` before the intended ContractError.
Check isinstance(kind, str) first.
- validation/runner.py: the build_from_columns csv_source guard was gated
on truthiness (`if csv_source and not isinstance(...)`), so csv_source:
0 / false / [] / {} skipped both the type check and the header check
below -- a malformed python-csv candidate could pass promotion silently.
Gate on presence of the key instead.
Signed-off-by: Simone Carolini <simonecarolini.sc@gmail.com>
The documented and CI "Tests (runtime)" command only deselected `image`, so the new minio-backed csv-reader/validation-runner integration tests (marked `integration`) stayed selected -- the standard pre-PR command from CONTRIBUTING.md now required Docker despite that doc saying Docker is only needed for the Postgres/Trino integration tests. Deselect `integration` there too, and add a dedicated step that runs them explicitly on the same runner (docker is available; minio self-provisions via the minio_container fixture's own `docker run`), mirroring how the Postgres/Trino adapters split their integration tier into a separate job. Nothing is silently skipped: the new step names both test files. Signed-off-by: Simone Carolini <simonecarolini.sc@gmail.com>
1 participant
Adds a third node kind, python-csv: a contract-only node (no user script) that loads a CSV from
s3://orhttps://into its declared warehouse table, validated at the release gate by an empty-table build plus a CSV header check.What's in it
csv_source.py— URI grammar (s3:///https://only), presence-only header rule,CsvSourceReaderport (dependency-free).csv_readers/— S3 (boto3, reusesmake_s3_client) and HTTPS (urllib, ranged) adapters +reader_forcomposition edge.kindfield with per-kind rules:python-csvforbidsscript, requiresreadsexactly{csv: <uri>};source_hash = sha256(uri),shared_code_hash = "".kindon every wire entry, uri-basedsource_hash, no closure/lint for csv.produce_csv+run_nodedispatch onnode.kind; conform/load path shared.csv_sourceinbuild_from_columns— ranged header fetch + presence check,csv_header_extra_columnswarning on extras.example_csv.yml, README node-kinds docs,boundary-contract.mdkindparagraph.Testing
Real backends, no stubs: minio (S3), a live stdlib HTTP server, real postgres. Full suite 459 passed / 0 skipped (3 image-marked deselected); ruff + mypy clean. Integration tests run in CI (
-m integrationafter the postgres stack is up).Wire-contract note
kindis additive with defaultpython-model;contract_versionstays1. Becausekindentersconfig_hash, existing python-model nodes re-version once on the v0.4.0 deploy — a deliberate one-time churn; manifest-controller fold-parity is preserved.Deploy order (do NOT merge out of sequence)
The continuo control-plane PR pins runtime image
v0.4.0. Merge this, tag/publishv0.4.0images, THEN merge the continuo PR.continuo-engine-contractstays pinned==0.7.0.Known follow-ups (non-blocking, filed for later)
extra_columns, defaultraise) — fix in the continuo spec/arch docs.🤖 Generated with Claude Code