mixle.data.sources.sql_source module

SQL connector – any SQLAlchemy URL (Postgres / MySQL / SQLite / …) into a DataSource.

Optional: requires sqlalchemy (pip install mixle[sql]). One connector covers every RDBMS. Unlike most connectors here, a SQL result set is not a cheap thing to hold in memory – a large query must not have to fit in the driver, or in mixle, before the first record reaches an encoder. See SqlCursorSource for the streaming, disposal, and replay contract (MXR-080-0064).

class SqlCursorSource(url, query, columns=None, *, structure=EXCHANGEABLE, schema=None, batch_size=_STREAM_BATCH_ROWS)[source]

Bases: object

A cursor-backed, bounded-memory DataSource over a SQL query.

Unlike LazySource (read a factory once, cache the result as a list – the right contract for a connector that is cheap to hold in memory), this source never builds a Python list of the full result set internally. records()/encode() open a fresh connection and pull rows off a streaming server-side cursor (stream_results=True, fetched batch_size at a time via SQLAlchemy’s yield_per) straight into the caller/encoder, so a query with more rows than fit in memory can still be consumed one bounded batch at a time (MXR-080-0064). .materialize() remains available as an explicit, honestly-named opt-in for callers who really do want the whole result set as a list (e.g. a small lookup table) – it is no longer what records()/encode() do by default.

Replay policy: each call to records()/encode()/materialize() re-executes the query from scratch, on a fresh connection. A live server-side cursor cannot be rewound, so “replay” here means “run the query again” (as re-running it by hand would), not “replay a cached snapshot” – if the underlying table changes between calls, later calls see the new data. Two calls are fully independent: nothing about this object is safe to share a single in-flight iterator across two consumers, but taking two separate records() calls is always fine, including concurrently.

Disposal: every connection/engine this object opens is disposed in a finally, deterministically – on normal exhaustion, on an exception raised while a row is being processed downstream (the exception unwinds the consuming frame, which drops the last reference to the generator; CPython finalizes it synchronously, running the pending finally), and on early abandonment (explicitly call .close() on the iterator returned by records() for an immediate, GC-independent close; letting it fall out of scope also works under CPython’s refcounting but is not a language guarantee). There is deliberately no object-level close()/context-manager pair here (contrast HDF5ArraySource): the replay-by-re-execute design means no connection is held between calls, so there is nothing to close while the object is idle.

Grouped (partially-exchangeable) structure: assigning whole groups to partitions needs to know a group is complete before it can be flushed, which a single forward pass can only bound in memory if the query already yields rows contiguously by group key (e.g. ORDER BY the group column) – this is required and enforced, not assumed: a group key reappearing after the stream has moved on to a different group raises ValueError rather than silently splitting the group across partitions.

Parameters:

  • url (str)

  • query (str)

  • columns (list[str] | None)

  • structure (SampleStructure)

  • schema (Schema | None)

  • batch_size (int)

records()[source]

Stream one full query execution’s rows. Call again to re-run the query (see class docstring).

Return type:

Iterable[Any]

materialize()[source]

Run the query and return every row as a list – an explicit opt-in that bypasses the bounded-memory path; use only when the result is known to be small enough to hold in memory.

Return type:

list[Any]

encode(encoder, num_chunks=1, chunk_size=None, batch_size=None)[source]

Partition and encode straight off the cursor, in bounded batches – never a full record list.

Strideable structures (everything but partially-exchangeable) round-robin record i into partition i % n_parts as it streams by, matching the records[k::n] striding partition_records() uses on a materialized list – bit-identical partitioning, just computed without ever holding records itself. Partially-exchangeable (grouped) structure streams whole groups round-robin, in first-encountered order, exactly as partition_records() does, but requires the query to already yield rows contiguous by group key (see class docstring).

Like _encode_indexable() (MXR-080-0056’s sibling fix), a partition with no rows contributes nothing to the returned list rather than an explicit (0, ...) entry, and a partition may legitimately contribute more than one entry (one per flushed batch, in read order) instead of exactly one – every existing caller sums/iterates/ concatenates over however many entries the list contains rather than assuming exactly num_chunks of them, so this is not a behavior change from their point of view.

Parameters:

  • encoder (Any)

  • num_chunks (int)

  • chunk_size (int | None)

  • batch_size (int | None)

Return type:

list[tuple[int, Any]]

read_sql(url, query, columns=None, *, structure=EXCHANGEABLE, schema=None, batch_size=_STREAM_BATCH_ROWS)[source]

Read the rows of query against the database at url into a cursor-backed, bounded-memory SqlCursorSource – see its docstring for the streaming, disposal, and replay contract.

Parameters:

  • url (str)

  • query (str)

  • columns (list[str] | None)

  • structure (SampleStructure)

  • schema (Schema | None)

  • batch_size (int)

Return type:

SqlCursorSource