bootstack.data.BaseDataSource#
- class bootstack.data.BaseDataSource(page_size=10)#
Bases:
ABCAbstract base class for datasource implementations.
Provides shared utilities and enforces the DataSourceProtocol interface through abstract methods. Subclasses implement storage-specific logic while inheriting common functionality.
- Parameters:
page_size (int) – Number of records per page (default: 10)
- page_size#
Current page size setting
Example
class RedisDataSource(BaseDataSource): def __init__(self, redis_client, page_size=10): super().__init__(page_size) self.redis = redis_client def load(self, records): pass # Redis-specific implementation def page(self, page=None): pass # Redis-specific implementation
- 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.
- abstractmethod delete(record_id)#
Delete record by ID.
- abstractmethod deselect(record_id)#
Mark record as unselected.
- abstractmethod 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).
- abstractmethod get(record_id)#
Retrieve single record by ID.
- abstractmethod has_next_page()#
Check if more pages exist after current page.
- Returns:
True if next page exists, False otherwise
- Return type:
- abstractmethod insert(record)#
Create new record and return its ID.
- abstractmethod is_selected(record_id)#
Check whether a record is currently selected.
- abstractmethod load(records)#
Load data records into the datasource.
- move(record_id, target_index)#
Reorder a record to a new position.
Default returns False (not supported). Subclasses that maintain an explicit ordering should override.
- abstractmethod next_page()#
Advance to next page and return its records.
- observe(condition=None, *order)#
Observe a live result set for a
where/orderquery.Returns a
Streamthat 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
observefor 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 viaon_changeand refetch only their visible window withpage/page_slice.- Parameters:
- Returns:
A
Streamof 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
Streamof coarse change events; chainmap/filter/debounceandlistento 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
Streaminstead.- Returns:
A
Streamwhenhandleris omitted, otherwise a cancellable subscription handle.- Return type:
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)
- abstractmethod order(*keys)#
Sort rows by one or more keys.
- abstractmethod page(page=None)#
Get records for specified page (or current page if None).
- abstractmethod page_slice(start_index, count)#
Get records by start index and count (respects filter/sort).
- abstractmethod 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/orderview 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
FileSourceConfigcontrolling encoding/delimiter/etc.
- abstractmethod select(record_id)#
Mark record as selected.
- abstractmethod select_all(current_page_only=False)#
Select all records (optionally only current page).
- abstractmethod selected(page=None)#
Get selected records, optionally paginated.
- abstractmethod update(record_id, updates)#
Update record fields by ID.