refactor: Extract large functions, improve exception handling, remove duplicates

## Large function extraction
- servers.py: Extract 8 _impl functions from register_server_tools (449 lines)
- certificates.py: Extract 7 _impl functions from register_certificate_tools (386 lines)
- MCP tool wrappers now delegate to module-level implementation functions

## Exception handling improvements
- Replace 11 broad `except Exception` with specific types
- health.py: (OSError, subprocess.SubprocessError)
- configuration.py: (HaproxyError, IOError, OSError, ValueError)
- servers.py: (IOError, OSError, ValueError)
- certificates.py: FileNotFoundError, (subprocess.SubprocessError, OSError)

## Duplicate code extraction
- Add parse_servers_state() to utils.py (replaces 4 duplicate parsers)
- Add disable_server_slot() to utils.py (replaces duplicate patterns)
- Update health.py, servers.py, domains.py to use new helpers

## Other improvements
- Add TypedDict types in file_ops.py and health.py
- Set file permissions (0o600) for sensitive files
- Update tests to use specific exception types

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
kaffa
2026-02-03 13:23:51 +09:00
parent e66c5ddc7f
commit 06ab47aca8
12 changed files with 891 additions and 723 deletions

View File

@@ -1,7 +1,58 @@
"""Utility functions for HAProxy MCP Server."""
from typing import Dict, Generator
from .config import StatField
from .config import StatField, StateField, STATE_MIN_COLUMNS
from .haproxy_client import haproxy_cmd_batch
def parse_servers_state(state_output: str) -> dict[str, dict[str, dict[str, str]]]:
"""Parse 'show servers state' output.
Args:
state_output: Raw output from HAProxy 'show servers state' command
Returns:
Nested dict: {backend: {server: {addr: str, port: str, state: str}}}
Example:
state = haproxy_cmd("show servers state")
parsed = parse_servers_state(state)
# parsed["pool_1"]["pool_1_1"] == {"addr": "10.0.0.1", "port": "8080", "state": "2"}
"""
result: dict[str, dict[str, dict[str, str]]] = {}
for line in state_output.split("\n"):
parts = line.split()
if len(parts) >= STATE_MIN_COLUMNS:
backend = parts[StateField.BE_NAME]
server = parts[StateField.SRV_NAME]
addr = parts[StateField.SRV_ADDR]
port = parts[StateField.SRV_PORT]
state = parts[StateField.SRV_OP_STATE] if len(parts) > StateField.SRV_OP_STATE else ""
if backend not in result:
result[backend] = {}
result[backend][server] = {
"addr": addr,
"port": port,
"state": state,
}
return result
def disable_server_slot(backend: str, server: str) -> None:
"""Disable a server slot (set to maint and clear address).
Args:
backend: Backend name (e.g., 'pool_1')
server: Server name (e.g., 'pool_1_1')
Raises:
HaproxyError: If HAProxy command fails
"""
haproxy_cmd_batch([
f"set server {backend}/{server} state maint",
f"set server {backend}/{server} addr 0.0.0.0 port 0"
])
def parse_stat_csv(stat_output: str) -> Generator[Dict[str, str], None, None]: