SqliteDataSource

SQLite-backed datasource with persistence and SQL-native filtering. Extends BaseDataSource.

For in-memory storage, see MemoryDataSource. See the DataSource Guide for usage examples.

Bases: BaseDataSource

SQLite-backed data manager with pagination, filtering, sorting, and CRUD operations.

Provides persistent storage using SQLite database with automatic schema inference and SQL-native filtering/sorting. Supports all operations defined in DataSourceProtocol.

Parameters:

Name	Type	Description	Default
`name`	`str`	Database file path or ":memory:" for in-memory database (default: ":memory:")	`':memory:'`
`page_size`	`int`	Number of records per page (default: 10)	`10`

Attributes:

Name	Type	Description
`conn`		SQLite database connection
`page_size`		Current page size setting

Example

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

Note

The database connection persists for the lifetime of the object
Close the connection explicitly with conn.close() if needed
Schema is inferred from first record's data types
An internal row-identity column (_bs_row_id) is added automatically as PRIMARY KEY
An internal selection column (_bs_selected) is added automatically for selection tracking
These internal columns are filtered out of the display column list automatically

init

__init__(name: str = ':memory:', page_size: int = 10)

Create SQLite datasource and set initial pagination state.

Parameters:

Name	Type	Description	Default
`name`	`str`	Database file path or ':memory:' for an in-memory database.	`':memory:'`
`page_size`	`int`	Number of records returned per page during pagination.	`10`

set_data

set_data(
    records: Union[
        Sequence[Primitive],
        Sequence[dict[str, Any]],
        Sequence[Sequence[Any]],
    ],
    column_keys: Optional[Sequence[str]] = None,
) -> "SqliteDataSource"

Load records into database, creating table with inferred schema.

Parameters:

Name	Type	Description	Default
`records`	`Union[Sequence[Primitive], Sequence[dict[str, Any]], Sequence[Sequence[Any]]]`	Sequence of dicts, primitives, or row sequences.	required
`column_keys`	`Optional[Sequence[str]]`	Optional column names when supplying row sequences (lists/tuples).	`None`

Returns:

Type	Description
`'SqliteDataSource'`	Self for method chaining

set_filter

set_filter(where_sql: str = '')

Apply SQL WHERE clause filter.

The fragment is interpolated into the query unmodified, so the caller is responsible for ensuring it is trusted/author-controlled. Do not pass strings built from end-user input — use parameterized queries directly via self.conn instead.

set_sort

set_sort(order_by_sql: str = '')

Apply SQL ORDER BY clause for sorting.

Same trust contract as set_filter: the fragment is interpolated verbatim, so it must be author-controlled, not user input.

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)

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).

get_distinct_values

get_distinct_values(
    column: str, limit: int = 1000
) -> List[Any]

Get distinct values for a column.

Parameters:

Name	Type	Description	Default
`column`	`str`	Column name to get distinct values from.	required
`limit`	`int`	Maximum number of distinct values to return.	`1000`

Returns:

Type	Description
`List[Any]`	List of distinct values sorted alphabetically.