Files
rag-manager/ingest_pipeline/cli/tui/screens/base.py
2025-09-21 01:38:47 +00:00

385 lines
13 KiB
Python

"""Base screen classes for common CRUD patterns."""
from __future__ import annotations
from typing import TYPE_CHECKING, Generic, TypeVar
from textual import work
from textual.app import ComposeResult
from textual.binding import Binding
from textual.containers import Container
from textual.screen import ModalScreen, Screen
from textual.widget import Widget
from textual.widgets import Button, DataTable, LoadingIndicator, Static
from typing_extensions import override
if TYPE_CHECKING:
from ..utils.storage_manager import StorageManager
T = TypeVar("T")
class BaseScreen(Screen[object]):
"""Base screen with common functionality."""
def __init__(
self,
storage_manager: StorageManager,
*,
name: str | None = None,
id: str | None = None,
classes: str | None = None,
**kwargs: object,
) -> None:
"""Initialize base screen."""
super().__init__(name=name, id=id, classes=classes)
# Ignore any additional kwargs to avoid type issues
self.storage_manager = storage_manager
class CRUDScreen(BaseScreen, Generic[T]):
"""Base class for Create/Read/Update/Delete operations."""
BINDINGS = [
Binding("ctrl+n", "create_item", "New"),
Binding("ctrl+e", "edit_item", "Edit"),
Binding("ctrl+d", "delete_item", "Delete"),
Binding("f5", "refresh", "Refresh"),
Binding("escape", "app.pop_screen", "Back"),
]
def __init__(
self,
storage_manager: StorageManager,
*,
name: str | None = None,
id: str | None = None,
classes: str | None = None,
**kwargs: object,
) -> None:
"""Initialize CRUD screen."""
super().__init__(storage_manager, name=name, id=id, classes=classes)
self.items: list[T] = []
self.selected_item: T | None = None
self.loading = False
@override
def compose(self) -> ComposeResult:
"""Compose CRUD screen layout."""
yield Container(
Static(self.get_title(), classes="screen-title"),
self.create_toolbar(),
self.create_list_view(),
LoadingIndicator(id="loading"),
classes="crud-container",
)
def get_title(self) -> str:
"""Get screen title."""
return "CRUD Operations"
def create_toolbar(self) -> Container:
"""Create action toolbar."""
return Container(
Button("📝 New", id="new_btn", variant="primary"),
Button("✏️ Edit", id="edit_btn", variant="default"),
Button("🗑️ Delete", id="delete_btn", variant="error"),
Button("🔄 Refresh", id="refresh_btn", variant="default"),
classes="toolbar",
)
def create_list_view(self) -> DataTable[str]:
"""Create list view widget."""
table = DataTable[str](id="items_table")
table.add_columns(*self.get_table_columns())
return table
def get_table_columns(self) -> list[str]:
"""Get table column headers."""
raise NotImplementedError("Subclasses must implement get_table_columns")
async def load_items(self) -> list[T]:
"""Load items from storage."""
raise NotImplementedError("Subclasses must implement load_items")
def item_to_row(self, item: T) -> list[str]:
"""Convert item to table row."""
raise NotImplementedError("Subclasses must implement item_to_row")
async def create_item_dialog(self) -> T | None:
"""Show create item dialog."""
raise NotImplementedError("Subclasses must implement create_item_dialog")
async def edit_item_dialog(self, item: T) -> T | None:
"""Show edit item dialog."""
raise NotImplementedError("Subclasses must implement edit_item_dialog")
async def delete_item(self, item: T) -> bool:
"""Delete item."""
raise NotImplementedError("Subclasses must implement delete_item")
def on_mount(self) -> None:
"""Initialize screen."""
self.query_one("#loading").display = False
self.refresh_items()
@work(exclusive=True)
async def refresh_items(self) -> None:
"""Refresh items list."""
self.set_loading(True)
try:
self.items = await self.load_items()
await self.update_table()
finally:
self.set_loading(False)
async def update_table(self) -> None:
"""Update table with current items."""
table = self.query_one("#items_table", DataTable)
table.clear()
for item in self.items:
row_data = self.item_to_row(item)
table.add_row(*row_data)
def set_loading(self, loading: bool) -> None:
"""Set loading state."""
self.loading = loading
loading_widget = self.query_one("#loading")
loading_widget.display = loading
def action_create_item(self) -> None:
"""Create new item."""
self.run_worker(self._create_item_worker())
def action_edit_item(self) -> None:
"""Edit selected item."""
if self.selected_item:
self.run_worker(self._edit_item_worker())
def action_delete_item(self) -> None:
"""Delete selected item."""
if self.selected_item:
self.run_worker(self._delete_item_worker())
def action_refresh(self) -> None:
"""Refresh items."""
self.refresh_items()
async def _create_item_worker(self) -> None:
"""Worker for creating items."""
item = await self.create_item_dialog()
if item:
self.refresh_items()
async def _edit_item_worker(self) -> None:
"""Worker for editing items."""
if self.selected_item:
item = await self.edit_item_dialog(self.selected_item)
if item:
self.refresh_items()
async def _delete_item_worker(self) -> None:
"""Worker for deleting items."""
if self.selected_item:
success = await self.delete_item(self.selected_item)
if success:
self.refresh_items()
class ListScreen(BaseScreen, Generic[T]):
"""Base for paginated list views."""
def __init__(
self,
storage_manager: StorageManager,
page_size: int = 20,
*,
name: str | None = None,
id: str | None = None,
classes: str | None = None,
**kwargs: object,
) -> None:
"""Initialize list screen."""
super().__init__(storage_manager, name=name, id=id, classes=classes)
self.page_size = page_size
self.current_page = 0
self.total_items = 0
self.items: list[T] = []
@override
def compose(self) -> ComposeResult:
"""Compose list screen layout."""
yield Container(
Static(self.get_title(), classes="screen-title"),
self.create_filters(),
self.create_list_view(),
self.create_pagination(),
LoadingIndicator(id="loading"),
classes="list-container",
)
def get_title(self) -> str:
"""Get screen title."""
raise NotImplementedError("Subclasses must implement get_title")
def create_filters(self) -> Container:
"""Create filter widgets."""
raise NotImplementedError("Subclasses must implement create_filters")
def create_list_view(self) -> Widget:
"""Create list view widget."""
raise NotImplementedError("Subclasses must implement create_list_view")
async def load_page(self, page: int, page_size: int) -> tuple[list[T], int]:
"""Load page of items."""
raise NotImplementedError("Subclasses must implement load_page")
def create_pagination(self) -> Container:
"""Create pagination controls."""
return Container(
Button("◀ Previous", id="prev_btn", variant="default"),
Static("Page 1 of 1", id="page_info"),
Button("Next ▶", id="next_btn", variant="default"),
classes="pagination",
)
def on_mount(self) -> None:
"""Initialize screen."""
self.query_one("#loading").display = False
self.load_current_page()
@work(exclusive=True)
async def load_current_page(self) -> None:
"""Load current page."""
self.set_loading(True)
try:
self.items, self.total_items = await self.load_page(self.current_page, self.page_size)
await self.update_list_view()
self.update_pagination_info()
finally:
self.set_loading(False)
async def update_list_view(self) -> None:
"""Update list view with current items."""
raise NotImplementedError("Subclasses must implement update_list_view")
def update_pagination_info(self) -> None:
"""Update pagination information."""
total_pages = max(1, (self.total_items + self.page_size - 1) // self.page_size)
current_page_display = self.current_page + 1
page_info = self.query_one("#page_info", Static)
page_info.update(f"Page {current_page_display} of {total_pages}")
prev_btn = self.query_one("#prev_btn", Button)
next_btn = self.query_one("#next_btn", Button)
prev_btn.disabled = self.current_page == 0
next_btn.disabled = self.current_page >= total_pages - 1
def set_loading(self, loading: bool) -> None:
"""Set loading state."""
loading_widget = self.query_one("#loading")
loading_widget.display = loading
def on_button_pressed(self, event: Button.Pressed) -> None:
"""Handle button presses."""
if event.button.id == "prev_btn" and self.current_page > 0:
self.current_page -= 1
self.load_current_page()
elif event.button.id == "next_btn":
total_pages = (self.total_items + self.page_size - 1) // self.page_size
if self.current_page < total_pages - 1:
self.current_page += 1
self.load_current_page()
class FormScreen(ModalScreen[T], Generic[T]):
"""Base for input forms with validation."""
BINDINGS = [
Binding("escape", "app.pop_screen", "Cancel"),
Binding("ctrl+s", "save", "Save"),
Binding("enter", "save", "Save"),
]
def __init__(
self,
item: T | None = None,
*,
name: str | None = None,
id: str | None = None,
classes: str | None = None,
**kwargs: object,
) -> None:
"""Initialize form screen."""
super().__init__(name=name, id=id, classes=classes)
# Ignore any additional kwargs to avoid type issues
self.item = item
self.is_edit_mode = item is not None
@override
def compose(self) -> ComposeResult:
"""Compose form layout."""
title = "Edit" if self.is_edit_mode else "Create"
yield Container(
Static(f"{title} {self.get_item_type()}", classes="form-title"),
self.create_form_fields(),
Container(
Button("💾 Save", id="save_btn", variant="success"),
Button("❌ Cancel", id="cancel_btn", variant="default"),
classes="form-actions",
),
classes="form-container",
)
def get_item_type(self) -> str:
"""Get item type name for title."""
raise NotImplementedError("Subclasses must implement get_item_type")
def create_form_fields(self) -> Container:
"""Create form input fields."""
raise NotImplementedError("Subclasses must implement create_form_fields")
def validate_form(self) -> tuple[bool, list[str]]:
"""Validate form data."""
raise NotImplementedError("Subclasses must implement validate_form")
def get_form_data(self) -> T:
"""Get item from form data."""
raise NotImplementedError("Subclasses must implement get_form_data")
def on_mount(self) -> None:
"""Initialize form."""
if self.is_edit_mode and self.item:
self.populate_form(self.item)
def populate_form(self, item: T) -> None:
"""Populate form with item data."""
raise NotImplementedError("Subclasses must implement populate_form")
def action_save(self) -> None:
"""Save form data."""
is_valid, errors = self.validate_form()
if is_valid:
try:
item = self.get_form_data()
self.dismiss(item)
except Exception as e:
self.show_validation_errors([str(e)])
else:
self.show_validation_errors(errors)
def show_validation_errors(self, errors: list[str]) -> None:
"""Show validation errors to user."""
# This would typically show a notification or update error display
pass
def on_button_pressed(self, event: Button.Pressed) -> None:
"""Handle button presses."""
if event.button.id == "save_btn":
self.action_save()
elif event.button.id == "cancel_btn":
self.dismiss(None)