openwebui-cli/openwebui_cli/config.py
Danny Stocker fbe6832d3e Improve CLI streaming, error handling, and add comprehensive tests
**Core Improvements:**
- Fix Pydantic v2 deprecation warning in config.py by migrating to ConfigDict
- Add graceful Ctrl-C handling for streaming chat responses
- Add connection error handling mid-stream with actionable suggestions
- Implement --history-file support for conversation context
- Enhance all error messages with specific troubleshooting steps

**Chat Command Enhancements:**
- Improved SSE streaming with KeyboardInterrupt handling
- Added --history-file option to load conversation history from JSON
- Support both array and object formats for history files
- Better error messages for missing prompts and invalid history files
- Graceful handling of connection timeouts during streaming

**Error Message Improvements:**
- 401: Clear authentication instructions with token expiry hints
- 403: Detailed permission error with possible causes
- 404: Resource not found with helpful suggestions
- 5xx: Server error with admin troubleshooting tips
- Network errors: Connection, timeout, and request failures with solutions

**Testing:**
- Add comprehensive test suite for chat commands (7 tests)
- Test streaming and non-streaming responses
- Test system prompts, stdin input, and JSON output
- Test conversation history loading
- Test RAG file and collection context
- All 14 tests passing

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-11-30 20:28:21 +01:00

111 lines
3.1 KiB
Python

"""Configuration management for OpenWebUI CLI."""
import os
from pathlib import Path
from typing import Any
import yaml
from pydantic import BaseModel, ConfigDict, Field
from pydantic_settings import BaseSettings
def get_config_dir() -> Path:
"""Get the configuration directory path (XDG-compliant)."""
if os.name == "nt": # Windows
base = Path(os.environ.get("APPDATA", Path.home() / "AppData" / "Roaming"))
else: # Unix/Linux/macOS
base = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config"))
return base / "openwebui"
def get_config_path() -> Path:
"""Get the configuration file path."""
return get_config_dir() / "config.yaml"
class ProfileConfig(BaseModel):
"""Configuration for a single server profile."""
uri: str = "http://localhost:8080"
# Token stored in keyring, not in config file
class DefaultsConfig(BaseModel):
"""Default settings for CLI commands."""
model: str | None = None
format: str = "text"
stream: bool = True
timeout: int = 30
class OutputConfig(BaseModel):
"""Output formatting preferences."""
colors: bool = True
progress_bars: bool = True
timestamps: bool = False
class Config(BaseModel):
"""Main configuration model."""
version: int = 1
default_profile: str = "default"
profiles: dict[str, ProfileConfig] = Field(default_factory=lambda: {"default": ProfileConfig()})
defaults: DefaultsConfig = Field(default_factory=DefaultsConfig)
output: OutputConfig = Field(default_factory=OutputConfig)
class Settings(BaseSettings):
"""Environment-based settings that override config file."""
model_config = ConfigDict(env_prefix="", case_sensitive=False)
openwebui_uri: str | None = None
openwebui_token: str | None = None
openwebui_profile: str | None = None
def load_config() -> Config:
"""Load configuration from file, with defaults for missing values."""
config_path = get_config_path()
if config_path.exists():
with open(config_path) as f:
data = yaml.safe_load(f) or {}
return Config(**data)
else:
return Config()
def save_config(config: Config) -> None:
"""Save configuration to file."""
config_path = get_config_path()
config_path.parent.mkdir(parents=True, exist_ok=True)
with open(config_path, "w") as f:
yaml.dump(config.model_dump(), f, default_flow_style=False, sort_keys=False)
def get_effective_config(
profile: str | None = None,
uri: str | None = None,
) -> tuple[str, str | None]:
"""
Get effective URI and profile name, respecting precedence:
CLI flags > env vars > config file > defaults
"""
config = load_config()
settings = Settings()
# Determine profile
effective_profile = profile or settings.openwebui_profile or config.default_profile
# Get profile config
profile_config = config.profiles.get(effective_profile, ProfileConfig())
# Determine URI with precedence
effective_uri = uri or settings.openwebui_uri or profile_config.uri
return effective_uri, effective_profile