MemoryDataSource

In-memory datasource with filtering, sorting, and pagination. Implements DataSourceProtocol.

For persistent storage, see SqliteDataSource. For file loading, see FileDataSource. See the DataSource Guide for usage examples.

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:

Name	Type	Description	Default
`page_size`	`int`	Number of records per page (default: 10)	`10`

Attributes:

Name	Type	Description
`page_size`		Current page size setting

Example

ds = MemoryDataSource(page_size=20)
ds.set_data([
    {"name": "Alice", "age": 30},
    {"name": "Bob", "age": 25},
])
ds.set_filter("age >= 30")
page = ds.get_page(0)

init

__init__(page_size: int = 10)

Initialize the in-memory datasource with defaults.

Parameters:

Name	Type	Description	Default
`page_size`	`int`	Number of records returned per page when paginating.	`10`

set_data

set_data(
    records: Union[
        Sequence[Primitive], Sequence[Dict[str, Any]]
    ],
) -> "MemoryDataSource"

Load records into datasource.

Parameters:

Name	Type	Description	Default
`records`	`Union[Sequence[Primitive], Sequence[Dict[str, Any]]]`	Sequence of dicts or primitives (auto-wrapped as {"text": str(x)})	required

Returns:

Type	Description
`'MemoryDataSource'`	Self for method chaining

set_filter

set_filter(where_sql: str = '')

Apply SQL-like WHERE filter to data.

set_sort

set_sort(order_by_sql: str = '')

Apply SQL-like ORDER BY sorting to data.

get_page

get_page(
    page: Optional[int] = None,
) -> List[Dict[str, Any]]

Get records for specified page.

next_page

next_page() -> List[Dict[str, Any]]

Advance to next page and return its records.

prev_page

prev_page() -> List[Dict[str, Any]]

Move to previous page and return its records.

has_next_page

has_next_page() -> bool

Check if more pages exist after current page.

total_count

total_count() -> int

Get total number of records matching current filter.

create_record

create_record(record: Dict[str, Any]) -> int

Create new record and return its ID.

read_record

read_record(record_id: Any) -> Optional[Dict[str, Any]]

Retrieve single record by ID.

update_record

update_record(
    record_id: Any, updates: Dict[str, Any]
) -> bool

Update record fields by ID.

delete_record

delete_record(record_id: Any) -> bool

Delete record by ID.

select_record

select_record(record_id: Any) -> bool

Mark record as selected.

unselect_record

unselect_record(record_id: Any) -> bool

Mark record as unselected.

select_all

select_all(current_page_only: bool = False) -> int

Select all records (optionally only current page).

unselect_all

unselect_all(current_page_only: bool = False) -> int

Unselect all records (optionally only current page).

get_selected

get_selected(
    page: Optional[int] = None,
) -> List[Dict[str, Any]]

Get selected records, optionally paginated.

selected_count

selected_count() -> int

Get total number of selected records.

export_to_csv

export_to_csv(
    filepath: str, include_all: bool = True
) -> None

Export records to CSV file.

get_page_from_index

get_page_from_index(
    start_index: int, count: int
) -> List[Dict[str, Any]]

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