bootstack.data.MemoryDataSource#

class bootstack.data.MemoryDataSource(page_size=10, id_field='id')#

Bases: BaseDataSource

In-memory data manager with pagination, filtering, sorting, and CRUD operations.

Stores all records in memory as dictionaries with automatic ID generation and selection tracking. Provides SQL-like filtering and sorting syntax for intuitive data manipulation.

The datasource maintains an internal index for O(1) ID lookups and supports dynamic schema inference from provided data.

Parameters:: page_size (int) – Number of records per page (default: 10)

page_size#: Current page size setting

Example

ds = MemoryDataSource(page_size=20)
ds.load([
    {"name": "Alice", "age": 30},
    {"name": "Bob", "age": 25},
])
ds.where(col("age") >= 30)
first = ds.page(0)

property count: int#: Total number of records matching the current filter.

property id_field: str#: Name of the record field that holds the stable row identity.

property selected_count: int#: Number of selected records.

close()#

Release any resources held by this data source.

The default is a no-op — in-memory sources hold none. Sources backed by a database connection or file handle override this to release it. Safe to call more than once.

delete(record_id)#

Delete record by ID.

deselect(record_id)#

Mark record as unselected.

deselect_all(current_page_only=False)#

Deselect all records (optionally only current page).

export_csv(filepath, include_all=True)#

Export records to a CSV file (streamed).

Parameters:

filepath (str) – Path to output CSV file
include_all (bool) – If True, export all records; if False, export only selected

get(record_id)#

Retrieve single record by ID.

has_next_page()#

Check if more pages exist after current page.

insert(record)#

Create new record and return its ID.

is_selected(record_id)#

Check whether a record is currently selected.

load(records)#

Load records into datasource.

Parameters:: records (Iterable[Primitive] | Iterable[Dict[str, Any]]) – An iterable of dicts or primitives (auto-wrapped as {“text”: str(x)}) — a list, a generator, or any other iterable.
Returns:: Self for method chaining
Return type:: MemoryDataSource

move(record_id, target_index)#

Reorder a record within the in-memory list.

next_page()#

Advance to next page and return its records.

observe(condition=None, *order)#

Observe a live result set for a where/order query.

Returns a Stream that emits the matching records immediately, then a fresh result set whenever a relevant change occurs. Each subscriber observes its own slice — declare the query once, react to its results over time (the “observable query” pattern).

Selection toggles do not re-emit (selection is not a row-set change). Unlike where/order, observing does not disturb the source’s own pagination view.

Performance: each relevant change re-runs the whole query and re-emits the full result set. Use observe for small derived sets — dashboard metrics, a short pinned list, a filtered side panel. For large or virtualized views (Table, ListView) do NOT observe the full set; bind those widgets to the source directly — they listen via on_change and refetch only their visible window with page/page_slice.

Parameters:

condition (Condition | None) – Filter condition built with col (or None for all rows).
order (str | Column | SortKey) – Sort keys — column names, "-name" for descending, or col(...) specs.

Returns:

A Stream of result sets (each a list of record dictionaries).

Return type:

Any

Example

ds.observe(col("status") == "active", "-created").listen(
    lambda rows: gauge.set_value(len(rows))
)

on_change(handler=None)#

Subscribe to changes to this source.

Call with no argument to get a composable Stream of coarse change events; chain map/filter/debounce and listen to drive any widget (for example, a dashboard badge bound to the row count). Call with a handler to subscribe directly and get back a cancellable subscription.

The handler receives a DataChangeEvent. Rapid mutations are coalesced into a single notification per event-loop turn, and mutations made from a background thread are delivered on the main thread automatically — so a bound widget can refresh from a worker-thread feed with no extra work.

Parameters:: handler (Callable[[Any], Any] | None) – Change handler. Omit to receive a Stream instead.
Returns:: A Stream when handler is omitted, otherwise a cancellable subscription handle.
Return type:: Any

Example

ds.on_change(lambda e: print("changed:", e.kind))

# Feed a dashboard badge with the live row count.
ds.on_change().map(lambda e: ds.count).listen(badge.set_value)

order(*keys)#

Sort rows by one or more keys (no arguments clears sorting).

page(page=None)#

Get records for specified page.

page_slice(start_index, count)#

Get records by start index and count (respects filter/sort).

prev_page()#

Move to previous page and return its records.

reload()#

Re-read data from the underlying source.

Default is a no-op suitable for in-memory implementations. File- and database-backed sources should override to re-query.

save(path, *, selected_only=False, format=None, config=None)#

Export records to a file, choosing the format by extension.

Records are streamed into the writer, so a large export does not materialize the whole dataset. The active where/order view is respected — what you export is what the source currently shows.

Parameters:

path (str) – Destination file path; its extension selects the format (CSV, TSV, JSON, JSONL, XML, and — with the extras — Parquet, Feather, HDF5).
selected_only (bool) – Export only selected records instead of all.
format (str | None) – Explicit format name overriding the path extension.
config (Any) – Optional FileSourceConfig controlling encoding/delimiter/etc.

select(record_id)#

Mark record as selected.

select_all(current_page_only=False)#

Select all records (optionally only current page).

selected(page=None)#

Get selected records, optionally paginated.

update(record_id, updates)#

Update record fields by ID.

where(condition=None)#

Filter rows by a condition built with col (None clears the filter).