"""Order Query Tool — query Azure Table Storage orders by userId via TUI.""" from __future__ import annotations import tomllib from dataclasses import dataclass from pathlib import Path from azure.data.tables import TableServiceClient from textual.app import App, ComposeResult from textual.containers import Horizontal, Vertical from textual.screen import ModalScreen from textual.widgets import Button, DataTable, Input, Static # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- CONFIG_DIR = Path.home() / ".order-query" CONFIG_FILE = CONFIG_DIR / "config.toml" @dataclass class Config: connection_string: str table_name: str = "order" def load_config() -> Config: if not CONFIG_FILE.exists(): CONFIG_DIR.mkdir(parents=True, exist_ok=True) CONFIG_FILE.write_text( "# Order Query Tool Configuration\n" 'connection_string = "DefaultEndpointsProtocol=https;AccountName=...;AccountKey=...;TableEndpoint=https://..."\n' "table_name = \"order\"\n" ) raise SystemExit( f"Config file created at {CONFIG_FILE}\n" "Please edit it with your Azure Storage connection string and run again." ) with open(CONFIG_FILE, "rb") as f: data = tomllib.load(f) return Config( connection_string=data["connection_string"], table_name=data.get("table_name", "order"), ) # --------------------------------------------------------------------------- # Order model # --------------------------------------------------------------------------- @dataclass class Order: order_id: str prod_id: str = "" prod_count: int = 0 prod_price: float = 0.0 revenue: float = 0.0 state: int = 0 order_time: str = "" receive_time: str = "" platform_order: str = "" custom_data: str = "" ver: int = 0 @property def state_text(self) -> str: return "Payed" if self.state == 0 else "Recived" # --------------------------------------------------------------------------- # Azure Table Storage client # --------------------------------------------------------------------------- def query_orders(config: Config, user_id: str) -> list[Order]: client = TableServiceClient.from_connection_string(config.connection_string) table = client.get_table_client(config.table_name) entities = table.query_entities( query_filter=f"PartitionKey eq '{user_id}'" ) orders: list[Order] = [] for entity in entities: orders.append( Order( order_id=entity.get("RowKey", ""), prod_id=entity.get("prodId", ""), prod_count=entity.get("prodCount", 0) or 0, prod_price=entity.get("prodPrice", 0.0) or 0.0, revenue=entity.get("revenue", 0.0) or 0.0, state=entity.get("state", 0), order_time=str(entity.get("OrderTime", "")), receive_time=str(entity.get("ReceiveTime", "")), platform_order=entity.get("platformOrder", ""), custom_data=entity.get("customData", ""), ver=entity.get("ver", 0) or 0, ) ) return orders # --------------------------------------------------------------------------- # TUI Screens # --------------------------------------------------------------------------- class InputScreen(ModalScreen[str]): """Screen for entering userId.""" def compose(self) -> ComposeResult: yield Vertical( Static("Enter User ID:", classes="label"), Input(placeholder="userId", id="user_id_input"), Horizontal( Button("Query", variant="primary", id="query_btn"), Button("Quit", variant="default", id="quit_btn"), classes="buttons", ), classes="input-container", ) def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "query_btn": user_id = self.query_one("#user_id_input", Input).value.strip() if user_id: self.dismiss(user_id) elif event.button.id == "quit_btn": self.app.exit() def on_input_submitted(self, event: Input.Submitted) -> None: user_id = event.value.strip() if user_id: self.dismiss(user_id) class OrderListScreen(ModalScreen[str | None]): """Screen showing order list as a DataTable.""" def __init__(self, orders: list[Order]) -> None: super().__init__() self.orders = orders def compose(self) -> ComposeResult: yield Vertical( Static(f"Found {len(self.orders)} orders. Select one for details."), DataTable(id="order_table"), Button("Back", id="back_btn"), classes="list-container", ) def on_mount(self) -> None: table = self.query_one("#order_table", DataTable) table.add_columns( "Order ID", "Product ID", "Count", "Price", "Revenue", "State", "Order Time", ) for order in self.orders: table.add_row( order.order_id, order.prod_id, str(order.prod_count), f"{order.prod_price:.2f}", f"{order.revenue:.2f}", order.state_text, order.order_time[:19] if order.order_time else "", key=order.order_id, ) def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: order_id = event.row_key.value self.dismiss(order_id) def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "back_btn": self.dismiss(None) class OrderDetailScreen(ModalScreen[None]): """Screen showing full order details.""" def __init__(self, order: Order) -> None: super().__init__() self.order = order def compose(self) -> ComposeResult: o = self.order detail_text = ( f"[bold]Order ID:[/bold] {o.order_id}\n" f"[bold]Product ID:[/bold] {o.prod_id}\n" f"[bold]Product Count:[/bold] {o.prod_count}\n" f"[bold]Product Price:[/bold] {o.prod_price:.2f}\n" f"[bold]Revenue:[/bold] {o.revenue:.2f}\n" f"[bold]State:[/bold] {o.state_text}\n" f"[bold]Order Time:[/bold] {o.order_time[:19] if o.order_time else 'N/A'}\n" f"[bold]Receive Time:[/bold] {o.receive_time[:19] if o.receive_time else 'N/A'}\n" f"[bold]Platform Order:[/bold] {o.platform_order or 'N/A'}\n" f"[bold]Version:[/bold] {o.ver}\n" f"[bold]Custom Data:[/bold] {o.custom_data or 'N/A'}" ) yield Vertical( Static(detail_text), Button("Back", id="back_btn"), classes="detail-container", ) def on_button_pressed(self, event: Button.Pressed) -> None: self.dismiss(None) # --------------------------------------------------------------------------- # Main App # --------------------------------------------------------------------------- class OrderQueryApp(App): """Main TUI application.""" CSS = """ .input-container { align: center middle; width: 60; height: auto; padding: 2; border: round $primary; } .label { text-align: center; margin-bottom: 1; text-style: bold; } .buttons { margin-top: 1; align: center middle; } .list-container { padding: 1; } .detail-container { align: center middle; width: 80; height: auto; padding: 2; border: round $primary; } #back_btn { margin-top: 1; } """ BINDINGS = [("q", "quit", "Quit")] _orders: list[Order] = [] def on_mount(self) -> None: self.push_screen(InputScreen(), self._handle_user_id) def _handle_user_id(self, user_id: str | None) -> None: if user_id is None: return try: config = load_config() self._orders = query_orders(config, user_id) except SystemExit: raise except Exception as e: self.notify(str(e), severity="error") self.push_screen(InputScreen(), self._handle_user_id) return if not self._orders: self.notify("No orders found for this user", severity="warning") self.push_screen(InputScreen(), self._handle_user_id) return self.push_screen(OrderListScreen(self._orders), self._handle_order_selected) def _handle_order_selected(self, order_id: str | None) -> None: if order_id is None: self.push_screen(InputScreen(), self._handle_user_id) return order = next((o for o in self._orders if o.order_id == order_id), None) if order: self.push_screen( OrderDetailScreen(order), lambda _: self._show_list_again() ) def _show_list_again(self) -> None: self.push_screen(OrderListScreen(self._orders), self._handle_order_selected) def main() -> None: app = OrderQueryApp() app.run() if __name__ == "__main__": main()