DataSourceProtocol
Protocol defining the interface for datasource implementations.
All datasources implement this protocol: MemoryDataSource, SqliteDataSource, FileDataSource. For a base implementation, see BaseDataSource.
Bases: Protocol
Protocol defining the interface for data source implementations.
This protocol establishes a contract that all datasource backends must implement, ensuring consistent behavior across different storage mechanisms (memory, database, web API, etc.).
All datasource implementations must support
- Pagination with configurable page size
- Filtering using SQL-like WHERE syntax
- Sorting using SQL-like ORDER BY syntax
- Full CRUD operations on records
- Selection tracking for multi-select scenarios
- CSV export capabilities
- Direct index-based data access
Attributes:
| Name | Type | Description |
|---|---|---|
page_size |
int
|
Number of records per page |
Notes
- Records are represented as Dict[str, Any] with at least 'id' and 'selected' fields
- Filtering and sorting syntax follows SQL conventions for familiarity
- All methods preserve immutability - operations return new data or modify in-place
set_data
set_data(
records: Sequence[Primitive]
| Sequence[Mapping[str, Any]],
) -> DataSourceProtocol
Load data records into the datasource.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
records
|
Sequence[Primitive] | Sequence[Mapping[str, Any]]
|
Sequence of records (dicts) or primitives (auto-wrapped) |
required |
Returns:
| Type | Description |
|---|---|
DataSourceProtocol
|
Self for method chaining |
set_filter
set_filter(where_sql: str = '') -> None
Apply SQL-like WHERE clause filter to data.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
where_sql
|
str
|
SQL WHERE condition (e.g., "age > 25 AND status = 'active'") |
''
|
set_sort
set_sort(order_by_sql: str = '') -> None
Apply SQL-like ORDER BY clause to data.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
order_by_sql
|
str
|
SQL ORDER BY clause (e.g., "name ASC, age DESC") |
''
|
get_page
get_page(page: Optional[int] = None) -> List[Record]
Get records for specified page (or current page if None).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
page
|
Optional[int]
|
Page number (0-indexed); updates current page if provided |
None
|
Returns:
| Type | Description |
|---|---|
List[Record]
|
List of record dictionaries for the page |
next_page
next_page() -> List[Record]
Advance to next page and return its records.
Returns:
| Type | Description |
|---|---|
List[Record]
|
List of record dictionaries for the new page |
prev_page
prev_page() -> List[Record]
Move to previous page and return its records.
Returns:
| Type | Description |
|---|---|
List[Record]
|
List of record dictionaries for the new page |
has_next_page
has_next_page() -> bool
Check if more pages exist after current page.
Returns:
| Type | Description |
|---|---|
bool
|
True if next page exists, False otherwise |
total_count
total_count() -> int
Get total number of records matching current filter.
Returns:
| Type | Description |
|---|---|
int
|
Total record count (respects active filter) |
create_record
create_record(record: Dict[str, Any]) -> int
Create new record and return its ID.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
record
|
Dict[str, Any]
|
Dictionary with record data |
required |
Returns:
| Type | Description |
|---|---|
int
|
The ID assigned to the new record |
read_record
read_record(record_id: Any) -> Optional[Record]
Retrieve single record by ID.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
record_id
|
Any
|
Unique identifier of the record |
required |
Returns:
| Type | Description |
|---|---|
Optional[Record]
|
Record dictionary or None if not found |
update_record
update_record(
record_id: Any, updates: Dict[str, Any]
) -> bool
Update record fields by ID.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
record_id
|
Any
|
Unique identifier of the record |
required |
updates
|
Dict[str, Any]
|
Dictionary with fields to update |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if record was updated, False if not found |
delete_record
delete_record(record_id: Any) -> bool
Delete record by ID.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
record_id
|
Any
|
Unique identifier of the record |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if record was deleted, False if not found |
select_record
select_record(record_id: Any) -> bool
Mark record as selected.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
record_id
|
Any
|
Unique identifier of the record |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if record was selected, False if not found |
unselect_record
unselect_record(record_id: Any) -> bool
Mark record as unselected.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
record_id
|
Any
|
Unique identifier of the record |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if record was unselected, False if not found |
select_all
select_all(current_page_only: bool = False) -> int
Select all records (optionally only current page).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
current_page_only
|
bool
|
If True, select only records on current page |
False
|
Returns:
| Type | Description |
|---|---|
int
|
Number of records selected |
unselect_all
unselect_all(current_page_only: bool = False) -> int
Unselect all records (optionally only current page).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
current_page_only
|
bool
|
If True, unselect only records on current page |
False
|
Returns:
| Type | Description |
|---|---|
int
|
Number of records unselected |
get_selected
get_selected(page: Optional[int] = None) -> List[Record]
Get selected records, optionally paginated.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
page
|
Optional[int]
|
Optional page number for paginating selected records |
None
|
Returns:
| Type | Description |
|---|---|
List[Record]
|
List of selected record dictionaries |
selected_count
selected_count() -> int
Get total number of selected records.
Returns:
| Type | Description |
|---|---|
int
|
Count of selected records |
export_to_csv
export_to_csv(
filepath: str, include_all: bool = True
) -> None
Export records to CSV file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filepath
|
str
|
Path to output CSV file |
required |
include_all
|
bool
|
If True, export all records; if False, export only selected |
True
|
get_page_from_index
get_page_from_index(
start_index: int, count: int
) -> List[Record]
Get records by start index and count (respects filter/sort).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
start_index
|
int
|
Starting record index |
required |
count
|
int
|
Number of records to retrieve |
required |
Returns:
| Type | Description |
|---|---|
List[Record]
|
List of record dictionaries |