BaseDataSource
Abstract base class for datasource implementations. Enforces DataSourceProtocol and provides shared utilities and hooks.
Extend this class to create custom datasources. See the DataSource Guide for examples.
Bases: ABC
Abstract 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:
| 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
class RedisDataSource(BaseDataSource):
def __init__(self, redis_client, page_size=10):
super().__init__(page_size)
self.redis = redis_client
def set_data(self, records):
pass # Redis-specific implementation
def get_page(self, page=None):
pass # Redis-specific implementation
__init__
__init__(page_size: int = 10)
Initialize base datasource with pagination settings.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
page_size
|
int
|
Number of records returned per page when paginating. |
10
|
set_data
abstractmethod
set_data(
records: Sequence[Primitive]
| Sequence[Mapping[str, Any]],
) -> "BaseDataSource"
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 |
|---|---|
'BaseDataSource'
|
Self for method chaining |
set_filter
abstractmethod
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
abstractmethod
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
abstractmethod
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
abstractmethod
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
abstractmethod
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
abstractmethod
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
abstractmethod
total_count() -> int
Get total number of records matching current filter.
Returns:
| Type | Description |
|---|---|
int
|
Total record count (respects active filter) |
create_record
abstractmethod
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
abstractmethod
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
abstractmethod
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
abstractmethod
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
abstractmethod
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
abstractmethod
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
abstractmethod
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
abstractmethod
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
abstractmethod
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
abstractmethod
selected_count() -> int
Get total number of selected records.
Returns:
| Type | Description |
|---|---|
int
|
Count of selected records |
export_to_csv
abstractmethod
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
abstractmethod
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 |