""" Redis tab — interactive Redis CLI with DB selector, command history, and output console. """ import threading import customtkinter as ctk from core.redis_client import RedisClient from core.i18n import t from core.icons import icon_text class RedisTab(ctk.CTkFrame): def __init__(self, master, store): super().__init__(master, fg_color="transparent") self.store = store self._current_alias: str | None = None self._client: RedisClient | None = None self._command_history: list[str] = [] self._history_index: int = -1 self._build_ui() def _build_ui(self): # ── Top bar: DB selector + stats ── top_frame = ctk.CTkFrame(self, fg_color="transparent") top_frame.pack(fill="x", padx=15, pady=(15, 5)) # DB selector db_label = ctk.CTkLabel(top_frame, text=t("redis_db"), anchor="w", font=ctk.CTkFont(size=12, weight="bold")) db_label.pack(side="left", padx=(0, 5)) self._db_var = ctk.StringVar(value="0") self._db_selector = ctk.CTkOptionMenu( top_frame, values=[str(i) for i in range(16)], variable=self._db_var, width=70, command=self._on_db_changed, ) self._db_selector.pack(side="left", padx=(0, 15)) # Keys count self._keys_label = ctk.CTkLabel(top_frame, text=t("redis_keys") + ": —", font=ctk.CTkFont(size=12), text_color="#9ca3af") self._keys_label.pack(side="left", padx=(0, 15)) # Memory usage self._memory_label = ctk.CTkLabel(top_frame, text=t("redis_memory") + ": —", font=ctk.CTkFont(size=12), text_color="#9ca3af") self._memory_label.pack(side="left") # ── Command input row ── cmd_frame = ctk.CTkFrame(self, fg_color="transparent") cmd_frame.pack(fill="x", padx=15, pady=5) prompt_label = ctk.CTkLabel(cmd_frame, text="redis>", font=ctk.CTkFont(family="Consolas", size=13), text_color="#ef4444") prompt_label.pack(side="left", padx=(0, 5)) self._cmd_entry = ctk.CTkEntry(cmd_frame, placeholder_text=t("redis_command_placeholder"), font=ctk.CTkFont(family="Consolas", size=13)) self._cmd_entry.pack(side="left", fill="x", expand=True, padx=(0, 10)) self._cmd_entry.bind("", lambda e: self._execute_command()) self._cmd_entry.bind("", self._history_up) self._cmd_entry.bind("", self._history_down) # ── Buttons row ── btn_frame = ctk.CTkFrame(self, fg_color="transparent") btn_frame.pack(fill="x", padx=15, pady=5) self._exec_btn = ctk.CTkButton(btn_frame, text=icon_text("execute", t("redis_execute")), width=100, command=self._execute_command) self._exec_btn.pack(side="left", padx=(0, 5)) self._info_btn = ctk.CTkButton(btn_frame, text=icon_text("info", "INFO"), width=80, fg_color="#6b7280", hover_color="#4b5563", command=lambda: self._run_quick("INFO")) self._info_btn.pack(side="left", padx=(0, 5)) self._dbsize_btn = ctk.CTkButton(btn_frame, text=icon_text("hash", "DBSIZE"), width=90, fg_color="#6b7280", hover_color="#4b5563", command=lambda: self._run_quick("DBSIZE")) self._dbsize_btn.pack(side="left", padx=(0, 5)) self._scan_btn = ctk.CTkButton(btn_frame, text=icon_text("search", "SCAN"), width=80, fg_color="#6b7280", hover_color="#4b5563", command=lambda: self._run_quick("SCAN 0 COUNT 100")) self._scan_btn.pack(side="left", padx=(0, 5)) self._clear_btn = ctk.CTkButton(btn_frame, text=icon_text("clear", t("redis_clear")), width=80, fg_color="#374151", hover_color="#1f2937", command=self._clear_output) self._clear_btn.pack(side="right") # ── Output console ── self._output = ctk.CTkTextbox(self, font=ctk.CTkFont(family="Consolas", size=12), state="disabled") self._output.pack(fill="both", expand=True, padx=15, pady=(5, 5)) # ── Status bar ── self._status_bar = ctk.CTkLabel(self, text=t("redis_disconnected"), anchor="w", font=ctk.CTkFont(size=11), text_color="#9ca3af") self._status_bar.pack(fill="x", padx=15, pady=(0, 10)) # ── Public API ── def set_server(self, alias: str | None): """Called when user selects a server in sidebar.""" if self._client: try: self._client.disconnect() except Exception: pass self._current_alias = alias self._client = None self._command_history.clear() self._history_index = -1 self._clear_output() if alias: self._set_status(t("redis_ready").format(alias=alias), "#22c55e") self._refresh_stats() else: self._set_status(t("redis_disconnected"), "#9ca3af") self._keys_label.configure(text=t("redis_keys") + ": —") self._memory_label.configure(text=t("redis_memory") + ": —") # ── Command execution ── def _execute_command(self): cmd = self._cmd_entry.get().strip() if not cmd: return if not self._current_alias: self._append_output(t("no_server_selected")) return # Add to history if not self._command_history or self._command_history[-1] != cmd: self._command_history.append(cmd) self._history_index = -1 self._cmd_entry.delete(0, "end") self._append_output(f"redis> {cmd}") self._set_buttons_state("disabled") db = int(self._db_var.get()) def _do(): try: client = self._get_client() client.select_db(db) result = client.execute(cmd) formatted = self._format_result(result) self.after(0, lambda: self._append_output(formatted)) except Exception as e: self.after(0, lambda: self._append_output(f"(error) {e}")) finally: self.after(0, lambda: self._set_buttons_state("normal")) self.after(0, self._refresh_stats) threading.Thread(target=_do, daemon=True).start() def _run_quick(self, cmd: str): """Execute a preset command.""" self._cmd_entry.delete(0, "end") self._cmd_entry.insert(0, cmd) self._execute_command() def _get_client(self) -> RedisClient: if self._client is None: server = self.store.get_server(self._current_alias) if not server: raise ConnectionError(f"Server '{self._current_alias}' not found") self._client = RedisClient(server) if not self._client.connect(): self._client = None raise ConnectionError(f"Cannot connect to Redis '{self._current_alias}'") return self._client # ── Stats refresh ── def _refresh_stats(self): if not self._current_alias: return def _do(): try: client = self._get_client() db = int(self._db_var.get()) client.select_db(db) # Прямые методы возвращают int и dict, не форматированные строки keys_count = client.dbsize() info = client.info("memory") memory = info.get("used_memory_human", "—") if info else "—" keys_text = f"{keys_count:,}" if keys_count is not None else "—" self.after(0, lambda: self._keys_label.configure( text=t("redis_keys") + f": {keys_text}")) self.after(0, lambda: self._memory_label.configure( text=t("redis_memory") + f": {memory}")) except Exception: pass threading.Thread(target=_do, daemon=True).start() def _on_db_changed(self, _value: str): self._refresh_stats() # ── History navigation ── def _history_up(self, _event): if not self._command_history: return "break" if self._history_index == -1: self._history_index = len(self._command_history) - 1 elif self._history_index > 0: self._history_index -= 1 self._cmd_entry.delete(0, "end") self._cmd_entry.insert(0, self._command_history[self._history_index]) return "break" def _history_down(self, _event): if not self._command_history: return "break" if self._history_index == -1: return "break" if self._history_index < len(self._command_history) - 1: self._history_index += 1 self._cmd_entry.delete(0, "end") self._cmd_entry.insert(0, self._command_history[self._history_index]) else: self._history_index = -1 self._cmd_entry.delete(0, "end") return "break" # ── Output helpers ── def _format_result(self, result) -> str: """Format Redis response for display.""" if result is None: return "(nil)" if isinstance(result, bytes): return result.decode("utf-8", errors="replace") if isinstance(result, int): return f"(integer) {result}" if isinstance(result, list): if not result: return "(empty list or set)" lines = [] for i, item in enumerate(result, 1): val = item.decode("utf-8", errors="replace") if isinstance(item, bytes) else str(item) lines.append(f"{i}) \"{val}\"") return "\n".join(lines) if isinstance(result, str): return result return str(result) def _append_output(self, text: str): self._output.configure(state="normal") self._output.insert("end", text + "\n") self._output.configure(state="disabled") self._output.see("end") def _clear_output(self): self._output.configure(state="normal") self._output.delete("1.0", "end") self._output.configure(state="disabled") def _set_status(self, text: str, color: str = "#9ca3af"): self._status_bar.configure(text=text, text_color=color) def _set_buttons_state(self, state: str): for btn in (self._exec_btn, self._info_btn, self._dbsize_btn, self._scan_btn): btn.configure(state=state)