mixle.data.sources.mongo_source module

MongoDB connector – a collection query into a DataSource.

Optional: requires pymongo (pip install mixle[mongo]). The _id field is dropped; supply a Schema to coerce the loosely-typed BSON documents. Like a SQL result set (see mixle.data.sources.sql_source), a MongoDB query result is not a cheap thing to hold in memory – a large collection query must not have to fit in the driver, or in mixle, before the first document reaches an encoder. See MongoCursorSource for the streaming, disposal, replay, and grouped-ordering contract (MXR-080-1666: “Mongo reads exhaust the cursor before yielding and never close the client” – independently numbered from, but the same architectural gap as, the SQL connector’s MXR-080-0064 fix: a different backend, the same “materialize the whole result before the first record is usable” defect).

class MongoCursorSource(uri, database, collection, query=None, columns=None, *, structure=EXCHANGEABLE, schema=None, batch_size=_STREAM_BATCH_DOCS, sort=None)[source]

Bases: object

A cursor-backed, bounded-memory DataSource over a MongoDB 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 client and pull documents off a batched cursor (find(...).batch_size(batch_size)) straight into the caller/encoder, so a collection query with more documents than fit in memory can still be consumed one bounded batch at a time (MXR-080-1666). .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 collection) – it is no longer what records()/encode() do by default.

Replay policy: each call to records()/encode()/materialize() re-runs the query from scratch, on a fresh MongoClient. A live 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 collection 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 client this object opens is closed in a finally, deterministically – on normal exhaustion, on an exception raised while a document 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). The live cursor itself is also closed explicitly (telling the server to free it immediately) before the client that owns it. 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.

Count resolution: a streaming cursor does not know its size up front, so a chunk_size target is resolved via count_documents() – one bounded integer, never the document data. Simpler than SqlCursorSource’s COUNT(*) subquery: count_documents takes the same filter dict _stream() already builds directly, with no query-text wrapping/parsing involved (there is no raw query string here to manipulate in the first place).

Projection: columns=None fetches whole documents (minus _id, always dropped); an explicitly empty columns=[] projects to zero columns (every record becomes ()), matching text_source.read_csv’s and the fixed hadoop_source.read_remote’s columns=[] semantics (MXR-080-0065’s sibling truthiness bug: if columns treats [] and None alike, which they are not) – implemented as a {"_id": 1} projection (the cheapest possible per-document payload), since an empty Mongo projection document {} means “no restriction” (returns full documents), not “zero fields”: the two only look alike under Python’s if columns truthiness, not in what either actually asks the server for. Requesting '_id' itself via columns is rejected at construction (it is unconditionally dropped, so honoring it would either silently do nothing or raise a confusing KeyError once the query actually runs).

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 documents contiguously by group key – 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. Unlike SqlCursorSource (which cannot inject an ORDER BY into a caller’s raw SQL text), this connector builds its query from structured pieces, so a plain string by is auto-sorted server-side (see _resolved_sort) – contiguous by construction, no caller action needed. A callable by has no server-side sort equivalent (Mongo cannot sort by an arbitrary Python function), so it still needs the caller to pass an explicit sort (or otherwise guarantee contiguous order) – exactly SQL’s ORDER BY-by-hand requirement, just for a narrower case. Either way, a string by also requires columns=None: a projected record is a tuple/scalar (positional), not a document dict, so SampleStructure.group_key cannot look a field up by name on it (mirrors SqlCursorSource, whose rows are always tuples and so always need a callable by) – rejected at construction rather than raising a confusing AttributeError deep in encode().

Parameters:

  • uri (str)

  • database (str)

  • collection (str)

  • query (dict | None)

  • columns (list[str] | None)

  • structure (SampleStructure)

  • schema (Schema | None)

  • batch_size (int)

  • sort (Any)

records()[source]

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

Return type:

Iterable[Any]

materialize()[source]

Run the query and return every document 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 documents contiguous by group key (see class docstring).

Like _encode_indexable() (MXR-080-0056’s sibling fix), a partition with no documents 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_mongo(uri, database, collection, query=None, columns=None, *, structure=EXCHANGEABLE, schema=None, batch_size=_STREAM_BATCH_DOCS, sort=None)[source]

Read documents matching query from database.collection at uri into a cursor-backed, bounded-memory MongoCursorSource – see its docstring for the streaming, disposal, replay, and grouped-ordering contract.

Parameters:

  • uri (str)

  • database (str)

  • collection (str)

  • query (dict | None)

  • columns (list[str] | None)

  • structure (SampleStructure)

  • schema (Schema | None)

  • batch_size (int)

  • sort (Any)

Return type:

MongoCursorSource