mixle.data.core module

DataSource – a lazy, typed, structured reference to data that yields encoder-ready records.

This is the single concept that replaces the “is it a list? a DataFrame? an RDD? a SQL cursor?” branching scattered across call sites. A DataSource carries a Schema (logical field types) and a SampleStructure (its exchangeability class), and knows how to yield records() and partition itself safely.

It is purely additive: seq_encode(list) and seq_encode(rdd) are untouched fast paths; seq_encode gains one branch that recognizes a DataSource and routes through its structure-aware encoder, returning the same [(count, payload)] shape consumers already expect.

class DataSource(*args, **kwargs)

Bases: Protocol

A lazy, typed, structured source of encoder-ready records.

records()

Yield raw records compatible with an encoder’s input type.

Return type:

Iterable[Any]

encode(encoder, num_chunks=1, chunk_size=None)

Partition (structure-aware) and seq_encode -> the same [(count, payload)] shape.

Parameters:

• encoder (Any)

• num_chunks (int)

• chunk_size (int | None)

Return type:

Any

class MaterializedSource(data, structure=EXCHANGEABLE, schema=None)

Bases: object

An in-memory DataSource wrapping a Sequence – what a bare list becomes.

Parameters:

• data (Sequence[Any])

• structure (SampleStructure)

• schema (Schema | None)

records()

Return conformed records from the owned replayable snapshot.

Return type:

Iterable[Any]

materialize()

Return a caller-owned list copy of the conformed snapshot.

Return type:

list[Any]

partition(n, *, by=None)

Split into n structure-safe sub-sources (group-aware for partially-exchangeable data).

Parameters:

• n (int)

• by (Any)

Return type:

list[MaterializedSource]

encode(encoder, num_chunks=1, chunk_size=None)

Encode the materialized records through structure-aware partitioning.

Parameters:

• encoder (Any)

• num_chunks (int)

• chunk_size (int | None)

Return type:

list[tuple[int, Any]]

class LazySource(factory, structure=EXCHANGEABLE, schema=None, length=None)

Bases: object

A DataSource that defers reading to a records factory and materializes on demand.

Connectors (Parquet, SQL, CSV, …) return one of these so open(...) does no I/O until the data is actually encoded; the records are read (and schema-coerced) once and cached.

Parameters:

• factory (Any)

• structure (SampleStructure)

• schema (Schema | None)

• length (int | None)

materialize()

Return a caller-owned list copy of the immutable, conformed cache.

Return type:

list[Any]

records()

Return an iterator over the cached materialized records.

Return type:

Iterable[Any]

partition(n, *, by=None)

Materialize and split into n structure-aware in-memory sources.

Parameters:

• n (int)

• by (Any)

Return type:

list[MaterializedSource]

encode(encoder, num_chunks=1, chunk_size=None)

Materialize records and encode them through structure-aware partitioning.

Parameters:

• encoder (Any)

• num_chunks (int)

• chunk_size (int | None)

Return type:

list[tuple[int, Any]]

as_source(data, structure=EXCHANGEABLE, schema=None)

Coerce data to a DataSource (pass a source through; wrap a sequence as materialized).

Parameters:

• data (Any)

• structure (SampleStructure)

• schema (Schema | None)

Return type:

DataSource

num_chunks_for(size, num_chunks=1, chunk_size=None)

Resolve the chunk count from an explicit num_chunks or a target chunk_size (as seq_encode does).

Both controls are validated as exact positive integers before any arithmetic: chunk_size=0 used to raise a bare ZeroDivisionError, and a negative/fractional chunk_size or a nonpositive/fractional num_chunks used to silently collapse to one chunk instead of failing – these bound memory, so silently weakening them can materially change an execution plan. A non-default num_chunks supplied together with chunk_size is rejected as conflicting rather than silently letting chunk_size win.

Parameters:

• size (int)

• num_chunks (int)

• chunk_size (int | None)

Return type:

int