Local format source boundary ============================ This page describes the development source tree. The published ``0.1.0.dev0`` package predates the local source boundary and CSV reader APIs described here. ``pipewhy.formats.verified_local_file(path, sha256_hex=digest, byte_limit=limit)`` opens a regular local file under an explicit lowercase SHA-256 content identity. The positive byte limit applies to the source and to reads through the returned binary handle. The context manager owns and closes the handle, checks the content and file identity before use, and verifies them again on successful exit. An exception in the caller propagates and closes the handle. A mismatched digest, changed file, replacement path, unsupported source or exceeded limit fails rather than reporting a successful read. The check detects accidental changes; it does not authenticate a producer or guarantee an atomic snapshot against an adversary. The expected digest must come from the caller's source-version evidence. ``LocalSourceError.reason`` distinguishes invalid configuration, unsupported sources, byte limits, version mismatches, changed sources and prematurely closed handles. Filesystem I/O errors remain explicit exceptions. This boundary does not parse a format or publish a snapshot. A capture coordinator must wait for complete parsing and successful context exit before reporting COMPLETE or publishing a snapshot. Remote object sources require separate adapters and their own version and consistency guarantees. The low-level CSV and JSON Lines readers also accept a ``SourceOpener``: a caller-supplied context manager factory receiving the expected digest and byte limit. The opener owns the binary stream and must enforce its size limit, verify the complete source against that digest, and check the source again before a successful exit. Reader batches remain provisional through that exit. A custom opener is a trusted boundary: the reader cannot prove that an implementation performed those checks. Snapshot adapters currently accept only the verified local-file path, and accepting an opener does not itself add a remote connector or establish S3-compatible storage support. CSV record batches ------------------ ``pipewhy.formats.iter_csv_batches`` reads an explicitly versioned local file using a ``CsvSchema`` and ``CsvLimits``. Schemas contain ordered ``CsvField`` definitions with stable field IDs, declared types and explicit null markers. The header must match the schema. Text, booleans, bounded signed and unsigned integers, finite decimals, finite binary64 values and supported timestamps are available. Strict normalization is the default; logical normalization must be selected explicitly. ``CsvSchema.descriptor`` returns the versioned schema description used to derive its SHA-256 identity. ``CsvSchema.from_descriptor`` validates and reconstructs that description, rejecting unknown fields, unsupported versions and invalid field definitions. Each returned description is independent of the schema object. The caller supplies limits for source bytes, record bytes, field bytes, field count, records and words per batch, and total records. An oversized or malformed record raises ``CsvReadError`` without including its value in the error message. Source failures retain their specific category in the CSV error reason. The interpreter's current CSV field limit can be inspected with ``csv_field_character_limit()``. The reader does not change that process-wide setting. Batches are provisional until the iterator is fully exhausted and source verification succeeds. Closing it early is not a complete capture. Remote storage remains separate work. Configured local CSV source --------------------------- ``pipewhy.adapters.csv.CsvSource`` binds a caller-selected logical source ID, local path, schema and limits. ``CsvAdapter`` discovers the configured root and records dataset without opening the file. Complete discovery therefore means the configured scope is enumerated; it does not assert that the file is readable. Reads require the exact lowercase SHA-256 of the requested file bytes. A requested batch size is an upper bound and is capped by the configured batch limit. The adapter does not retain historical versions; a caller may supply an older version only while its exact bytes are still available at the configured path. Source lineage remains UNKNOWN with an explicit reason. Emitted batches remain provisional until the iterator completes and verifies its source. CSV snapshots ------------- The Python ``scan_source`` API accepts a ``CsvSource`` and an exact source digest. Once a complete read and verification succeed, the capture records the number of rows actually observed as both its expected and returned counts; this is not an independent count supplied by the upstream producer. The capture stores the schema descriptor and can be reopened without the source file. Failures discard provisional records and publish an unavailable observation with an explicit reason. Source lineage remains unknown. Command-line configuration -------------------------- Select CSV explicitly and supply the schema and resource limits in the TOML configuration. Loading the configuration does not open the source file:: schema_version = 1 [store] path = "snapshots" [sources.orders] kind = "csv" path = "orders.csv" source_id = "orders" schema = { format = "pipewhy.csv-schema", version = 1, encoding = "utf-8", fields = [{ id = 0, name = "amount", type = "int64", nullable = false, null_markers = [] }] } limits = { source_bytes = 4096, record_bytes = 1024, field_bytes = 256, field_count = 16, batch_records = 2, batch_words = 1024, max_records = 100 } The source version for CSV is the full lowercase SHA-256 digest of the exact file bytes. For example, obtain it with ``python -c "import hashlib; print(hashlib.sha256(open('orders.csv','rb').read()).hexdigest())"`` and pass the result to ``pipewhy scan orders --config pipewhy.toml --source-version DIGEST``. The example digest command reads the whole file; for large files, calculate the same digest with a streaming checksum tool. PipeWhy verifies the digest while reading and never assumes a filename implies the format. CSV support uses the Python standard library. JSON Lines record batches ------------------------- The development-source API ``iter_jsonl_batches`` reads explicitly configured flat JSON objects, one per UTF-8 line, into provisional normalized batches. A ``JsonlSchema`` declares stable field IDs, scalar types, required fields and nullability; absent optional fields and present nulls remain distinct. The reader rejects duplicate or unknown fields, nested data, malformed input and values outside its explicit limits. Its source digest and batches have the same local verification and provisional-read rules described above. The line limit counts raw bytes including any line terminator; the field-byte limit counts UTF-8 bytes in decoded field names and string values. Numeric token length and decimal exponent have separate explicit limits. ``JsonlSchema.descriptor`` returns a fresh versioned schema description, and ``JsonlSchema.from_descriptor`` validates and reconstructs it. The schema identity includes field IDs, names, types, presence and null policies. ``JsonlSource`` and ``JsonlAdapter`` bind a caller-selected logical source ID, local path, schema and limits. Discovery enumerates only the configured scope and does not establish that the file is readable. Reads use the exact SHA-256 digest, and upstream lineage is unknown. This adapter does not retain historical files or publish snapshots by itself. The Python ``scan_source`` API accepts a ``JsonlSource``, stores the schema descriptor and reopens the capture without the original file. Complete counts are observed after verification, not an independent upstream count. Failed reads discard provisional records and retain an explicit reason. For command-line scans, select ``kind = "jsonl"`` in a source table. For example:: [sources.events] kind = "jsonl" path = "events.jsonl" source_id = "events" schema = { format = "pipewhy.jsonl-schema", version = 1, fields = [{ id = 0, name = "id", type = "int64", required = true, nullable = false }, { id = 1, name = "note", type = "text", required = false, nullable = true }] } limits = { source_bytes = 4096, line_bytes = 1024, field_bytes = 256, field_count = 8, batch_records = 2, batch_words = 1024, max_records = 100, number_bytes = 32, decimal_exponent = 100 } Include the same top-level ``schema_version`` and store configuration shown for CSV, then run ``pipewhy scan events --config pipewhy.toml --source-version DIGEST`` with the full lowercase SHA-256 of the exact JSON Lines file bytes. The format is selected by configuration, never inferred from its file extension. JSON document record batches ---------------------------- ``iter_json_document_batches`` accepts a ``JsonDocumentSchema`` with explicit ``object`` or ``array`` framing and a ``JsonDocumentLimits``. An object is one record; an array contains flat objects in source order. The initial subset uses the same declared scalar, null and missing-field policies as JSON Lines. Nested values, duplicate or unknown keys, malformed documents and mismatched framing fail explicitly. The complete document is read into memory under an explicit document-byte bound, separate from the source-byte bound. Other limits cover fields, numeric tokens, decimal exponent, record count and batch size. A custom source opener must provide a blocking binary stream and fulfill the verification obligations above. All batches remain provisional until source verification completes. ``JsonDocumentSchema.descriptor`` includes the framing and field policies used to calculate its identity. ``JsonDocumentSchema.from_descriptor`` strictly validates and reconstructs this description for later capture metadata. ``JsonDocumentSource`` and ``JsonDocumentAdapter`` configure a local path, logical source ID, schema and limits. Discovery covers the configured scope without claiming the file is readable; source lineage remains unknown. Reads require an exact content digest and do not retain old file versions. The Python ``scan_source`` API accepts ``JsonDocumentSource`` and stores the framing and schema descriptor in a snapshot that can be reopened without the original file. Complete counts describe records observed after verification, not independent upstream counts. Failed reads discard provisional data and retain a reason. For command-line scans, select ``kind = "json-document"`` and declare the document framing explicitly:: [sources.orders] kind = "json-document" path = "orders.json" source_id = "orders" schema = { format = "pipewhy.json-document-schema", version = 1, framing = "array", fields = [{ id = 0, name = "amount", type = "int64", required = true, nullable = false }] } limits = { source_bytes = 4096, document_bytes = 4096, field_bytes = 256, field_count = 8, batch_records = 2, batch_words = 1024, max_records = 100, number_bytes = 32, decimal_exponent = 100 } With the top-level store configuration shown above, pass the full SHA-256 digest to ``pipewhy scan orders --config pipewhy.toml --source-version DIGEST``. Parquet input boundaries ------------------------ ``read_parquet_batches`` also accepts a trusted verified source opener. The reader stages its binary stream locally and independently checks its size and SHA-256 digest before starting the bounded worker. A failure in the opener's final verification prevents any decoded records from being returned. This interface allows storage adapters to supply verified streams; it does not itself implement a remote storage connector. Snapshot capture currently accepts local paths.