Skip to content

Commit 28169e4

Browse files
committed
refactor: address CodeFlow warnings (code duplication, complexity, style)
High priority: - Eliminate code duplication between _fact_extractors.py and test_doc_sync.py by importing shared extractors (24 CodeDuplication/duplicate-code warnings) - Extract _build_enum_spec from _build_enriched_spec to reduce McCabe complexity from 14 to 8 - Extract _check_local_backend, _build_backends, _build_models, _pick_default_model, _pick_default_backend from sqlseed_list_gemma_models to reduce McCabe complexity from 13 to 3 - Extract _extract_func_name from get_mcp_tool_names to reduce nesting from 6/5 to 4/5 Low/Medium priority: - _hardware.py: add encoding='utf-8' to open(), use set for membership test, remove overlapping except (json.JSONDecodeError is subclass of ValueError), use NamedTuple for MODEL_REQUIREMENTS - server.py: use set for membership test in _pick_default_model - sync_docs.py: add E402 noqa for sys.path-dependent import - test_doc_sync.py: add E402 noqa for sys.path-dependent import Note: hookspecs.py unused-argument warnings are pluggy framework convention (placeholder parameters) and cannot be suppressed via ruff noqa since ARG rules are not enabled. These are informational only.
1 parent c1cb387 commit 28169e4

6 files changed

Lines changed: 224 additions & 359 deletions

File tree

plugins/mcp-server-sqlseed/src/mcp_server_sqlseed/server.py

Lines changed: 114 additions & 95 deletions
Original file line numberDiff line numberDiff line change
@@ -328,142 +328,161 @@ def sqlseed_gemma4_agent_fill(
328328
}
329329

330330

331-
@mcp.tool()
332-
def sqlseed_list_gemma_models() -> dict[str, Any]:
333-
"""List Gemma 4 models with hardware compatibility and backend availability.
331+
_BACKEND_DESCRIPTIONS: dict[str, str] = {
332+
"google_ai_studio": "Google AI Studio API (free tier available, recommended)",
333+
"lm_studio": "LM Studio local deployment (http://127.0.0.1:1234, GUI-based)",
334+
"ollama": "Ollama local deployment (offline, CLI-based)",
335+
"openai_compat": "Any OpenAI-compatible API endpoint",
336+
}
337+
338+
_LOCAL_BACKEND_URLS: dict[str, str] = {
339+
"lm_studio": "http://127.0.0.1:1234/v1/models",
340+
"ollama": "http://localhost:11434/v1/models",
341+
}
342+
343+
_STATUS_ICONS: dict[str, str] = {
344+
"recommended": "recommended",
345+
"capable": "capable (meets minimum specs)",
346+
"capable_slow": "capable but likely slow (VRAM < minimum, will use RAM offloading)",
347+
"cpu_only": "CPU-only inference (no GPU detected)",
348+
"insufficient": "insufficient hardware",
349+
"cloud_only": "cloud API only",
350+
}
351+
352+
353+
def _check_local_backend(backend_id: str, url: str) -> dict[str, Any]:
354+
"""Check reachability and loaded models for a local LLM backend."""
355+
reachable = False
356+
loaded: list[str] = []
357+
try:
358+
req = urllib.request.Request(url)
359+
with urllib.request.urlopen(req, timeout=3) as resp:
360+
data = json.loads(resp.read().decode())
361+
loaded = [m.get("id", "unknown") for m in data.get("data", []) if m.get("id")]
362+
reachable = True
363+
except (OSError, ValueError):
364+
pass
365+
366+
if reachable and loaded:
367+
reason = f"{len(loaded)} model(s) loaded"
368+
elif reachable:
369+
reason = "Service running, no models loaded"
370+
else:
371+
reason = "Service not running"
334372

335-
Dynamically detects the current hardware environment (RAM, GPU/VRAM)
336-
and checks which LLM backends are reachable. Returns models annotated
337-
with compatibility status and backends annotated with availability.
338-
"""
339-
backend_descriptions: dict[str, str] = {
340-
"google_ai_studio": "Google AI Studio API (free tier available, recommended)",
341-
"lm_studio": "LM Studio local deployment (http://127.0.0.1:1234, GUI-based)",
342-
"ollama": "Ollama local deployment (offline, CLI-based)",
343-
"openai_compat": "Any OpenAI-compatible API endpoint",
373+
return {
374+
"id": backend_id,
375+
"description": _BACKEND_DESCRIPTIONS[backend_id],
376+
"available": reachable and bool(loaded),
377+
"reachable": reachable,
378+
"loaded_models": loaded,
379+
"reason": reason,
344380
}
345381

346-
if not _AI_AVAILABLE:
347-
return {
348-
"models": [],
349-
"backends": [
350-
{"id": bid, "description": desc, "available": False} for bid, desc in backend_descriptions.items()
351-
],
352-
"hardware": {},
353-
"error": "sqlseed-ai plugin not installed. Install with: pip install sqlseed-ai",
354-
}
355-
356-
# ── 1. Detect hardware ──
357-
hw = detect_hardware()
358382

359-
# ── 2. Check backend availability ──
360-
ai_config = AIConfig.from_env()
361-
backends_result = []
383+
def _build_backends(ai_config: Any) -> list[dict[str, Any]]:
384+
"""Build the list of backend availability info."""
385+
backends: list[dict[str, Any]] = []
362386

363387
# Google AI Studio: check API key
364388
has_api_key = ai_config.has_real_api_key
365-
backends_result.append(
389+
backends.append(
366390
{
367391
"id": "google_ai_studio",
368-
"description": backend_descriptions["google_ai_studio"],
392+
"description": _BACKEND_DESCRIPTIONS["google_ai_studio"],
369393
"available": has_api_key,
370394
"reason": "API key configured" if has_api_key else "No API key (set GOOGLE_API_KEY or SQLSEED_AI_API_KEY)",
371395
}
372396
)
373397

374398
# LM Studio / Ollama: check service reachability + loaded models
375-
local_urls: dict[str, str] = {
376-
"lm_studio": "http://127.0.0.1:1234/v1/models",
377-
"ollama": "http://localhost:11434/v1/models",
378-
}
379-
for backend_id, url in local_urls.items():
380-
reachable = False
381-
loaded: list[str] = []
382-
try:
383-
req = urllib.request.Request(url)
384-
with urllib.request.urlopen(req, timeout=3) as resp:
385-
data = json.loads(resp.read().decode())
386-
loaded = [m.get("id", "unknown") for m in data.get("data", []) if m.get("id")]
387-
reachable = True
388-
except (OSError, ValueError):
389-
pass
390-
391-
if reachable and loaded:
392-
reason = f"{len(loaded)} model(s) loaded"
393-
elif reachable:
394-
reason = "Service running, no models loaded"
395-
else:
396-
reason = "Service not running"
397-
398-
backends_result.append(
399-
{
400-
"id": backend_id,
401-
"description": backend_descriptions[backend_id],
402-
"available": reachable and bool(loaded),
403-
"reachable": reachable,
404-
"loaded_models": loaded,
405-
"reason": reason,
406-
}
407-
)
399+
for backend_id, url in _LOCAL_BACKEND_URLS.items():
400+
backends.append(_check_local_backend(backend_id, url))
408401

409402
# OpenAI-compatible: informational only
410-
backends_result.append(
403+
backends.append(
411404
{
412405
"id": "openai_compat",
413-
"description": backend_descriptions["openai_compat"],
406+
"description": _BACKEND_DESCRIPTIONS["openai_compat"],
414407
"available": False,
415408
"reason": "Requires explicit base_url configuration",
416409
}
417410
)
411+
return backends
418412

419-
# ── 3. Build model list with compatibility status ──
420-
status_icons: dict[str, str] = {
421-
"recommended": "recommended",
422-
"capable": "capable (meets minimum specs)",
423-
"capable_slow": "capable but likely slow (VRAM < minimum, will use RAM offloading)",
424-
"cpu_only": "CPU-only inference (no GPU detected)",
425-
"insufficient": "insufficient hardware",
426-
"cloud_only": "cloud API only",
427-
}
428413

414+
def _build_models(hw: dict[str, Any]) -> list[dict[str, Any]]:
415+
"""Build the list of Gemma models with hardware compatibility status."""
429416
models = []
430417
for member in GemmaModel:
431418
status = evaluate_model_status(member.value, hw)
432-
model_req = MODEL_REQUIREMENTS.get(member.value, {})
419+
req = MODEL_REQUIREMENTS.get(member.value)
433420
models.append(
434421
{
435422
"id": member.value,
436423
"display_name": member.display_name,
437424
"status": status,
438-
"status_description": status_icons.get(status, status),
425+
"status_description": _STATUS_ICONS.get(status, status),
439426
"local_only": member.is_local_only,
440427
"requirements": {
441-
"min_ram_gb": model_req.get("min_ram_gb", 0),
442-
"min_vram_gb": model_req.get("min_vram_gb", 0),
443-
"recommended_vram_gb": model_req.get("recommended_vram_gb", 0),
428+
"min_ram_gb": req.min_ram_gb if req else 0,
429+
"min_vram_gb": req.min_vram_gb if req else 0,
430+
"recommended_vram_gb": req.recommended_vram_gb if req else 0,
444431
},
445432
}
446433
)
434+
return models
435+
447436

448-
# ── 4. Determine best default ──
449-
# Pick the largest capable model (iterate from largest to smallest)
450-
default_model = GemmaModel.GEMMA_4_26B_A4B.value
437+
def _pick_default_model(models: list[dict[str, Any]]) -> str:
438+
"""Pick the largest capable model (iterate from largest to smallest)."""
451439
for m in reversed(models):
452-
if m["status"] in ("recommended", "capable") and not m["local_only"]:
453-
default_model = str(m["id"])
454-
break
455-
456-
# Pick the first available backend (prefer local over cloud)
457-
default_backend = "google_ai_studio"
458-
backend_priority = ["lm_studio", "ollama", "google_ai_studio", "openai_compat"]
459-
for b_id in backend_priority:
460-
for b in backends_result:
440+
if m["status"] in {"recommended", "capable"} and not m["local_only"]:
441+
return str(m["id"])
442+
return GemmaModel.GEMMA_4_26B_A4B.value
443+
444+
445+
def _pick_default_backend(backends: list[dict[str, Any]]) -> str:
446+
"""Pick the first available backend, preferring local over cloud."""
447+
priority = ["lm_studio", "ollama", "google_ai_studio", "openai_compat"]
448+
for b_id in priority:
449+
for b in backends:
461450
if b["id"] == b_id and b.get("available"):
462-
default_backend = b_id
463-
break
464-
else:
465-
continue
466-
break
451+
return b_id
452+
return "google_ai_studio"
453+
454+
455+
@mcp.tool()
456+
def sqlseed_list_gemma_models() -> dict[str, Any]:
457+
"""List Gemma 4 models with hardware compatibility and backend availability.
458+
459+
Dynamically detects the current hardware environment (RAM, GPU/VRAM)
460+
and checks which LLM backends are reachable. Returns models annotated
461+
with compatibility status and backends annotated with availability.
462+
"""
463+
if not _AI_AVAILABLE:
464+
return {
465+
"models": [],
466+
"backends": [
467+
{"id": bid, "description": desc, "available": False} for bid, desc in _BACKEND_DESCRIPTIONS.items()
468+
],
469+
"hardware": {},
470+
"error": "sqlseed-ai plugin not installed. Install with: pip install sqlseed-ai",
471+
}
472+
473+
# ── 1. Detect hardware ──
474+
hw = detect_hardware()
475+
476+
# ── 2. Check backend availability ──
477+
ai_config = AIConfig.from_env()
478+
backends_result = _build_backends(ai_config)
479+
480+
# ── 3. Build model list with compatibility status ──
481+
models = _build_models(hw)
482+
483+
# ── 4. Determine best defaults ──
484+
default_model = _pick_default_model(models)
485+
default_backend = _pick_default_backend(backends_result)
467486

468487
return {
469488
"models": models,

plugins/sqlseed-ai/src/sqlseed_ai/_hardware.py

Lines changed: 24 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@
1313
import platform
1414
import subprocess
1515
import time
16-
from typing import Any
16+
from typing import Any, NamedTuple
1717

1818
from sqlseed._utils.logger import get_logger
1919

@@ -67,11 +67,11 @@ class MEMORYSTATUSEX(ctypes.Structure):
6767
def _get_ram_linux() -> tuple[float, float] | None:
6868
"""Get RAM from /proc/meminfo. Returns (total_gb, available_gb)."""
6969
try:
70-
with open("/proc/meminfo") as f:
70+
with open("/proc/meminfo", encoding="utf-8") as f:
7171
info: dict[str, int] = {}
7272
for line in f:
7373
parts = line.split()
74-
if parts[0] in ("MemTotal:", "MemAvailable:"):
74+
if parts[0] in {"MemTotal:", "MemAvailable:"}:
7575
info[parts[0].rstrip(":")] = int(parts[1]) # in kB
7676
total = info.get("MemTotal", 0) / (1024**2)
7777
avail = info.get("MemAvailable", 0) / (1024**2)
@@ -214,7 +214,7 @@ def _detect_gpu_macos() -> list[dict[str, Any]]:
214214
}
215215
)
216216
return gpus
217-
except (FileNotFoundError, json.JSONDecodeError, subprocess.TimeoutExpired, ValueError):
217+
except (FileNotFoundError, ValueError, subprocess.TimeoutExpired):
218218
return []
219219

220220

@@ -275,32 +275,22 @@ def detect_hardware() -> dict[str, Any]:
275275
# ── Model requirements ───────────────────────────────────────────────
276276

277277
# Approximate requirements for Gemma 4 (Q4_K_M quantized for local inference)
278-
MODEL_REQUIREMENTS: dict[str, dict[str, Any]] = {
279-
"gemma-4-e2b-it": {
280-
"min_ram_gb": 4,
281-
"min_vram_gb": 2,
282-
"recommended_vram_gb": 4,
283-
},
284-
"gemma-4-e4b-it": {
285-
"min_ram_gb": 6,
286-
"min_vram_gb": 3,
287-
"recommended_vram_gb": 6,
288-
},
289-
"gemma-4-12b-it": {
290-
"min_ram_gb": 12,
291-
"min_vram_gb": 8,
292-
"recommended_vram_gb": 10,
293-
},
294-
"gemma-4-26b-a4b-it": {
295-
"min_ram_gb": 16,
296-
"min_vram_gb": 14,
297-
"recommended_vram_gb": 16,
298-
},
299-
"gemma-4-31b-it": {
300-
"min_ram_gb": 24,
301-
"min_vram_gb": 18,
302-
"recommended_vram_gb": 24,
303-
},
278+
279+
280+
class ModelRequirement(NamedTuple):
281+
"""Hardware requirements for a local model variant."""
282+
283+
min_ram_gb: int
284+
min_vram_gb: int
285+
recommended_vram_gb: int
286+
287+
288+
MODEL_REQUIREMENTS: dict[str, ModelRequirement] = {
289+
"gemma-4-e2b-it": ModelRequirement(min_ram_gb=4, min_vram_gb=2, recommended_vram_gb=4),
290+
"gemma-4-e4b-it": ModelRequirement(min_ram_gb=6, min_vram_gb=3, recommended_vram_gb=6),
291+
"gemma-4-12b-it": ModelRequirement(min_ram_gb=12, min_vram_gb=8, recommended_vram_gb=10),
292+
"gemma-4-26b-a4b-it": ModelRequirement(min_ram_gb=16, min_vram_gb=14, recommended_vram_gb=16),
293+
"gemma-4-31b-it": ModelRequirement(min_ram_gb=24, min_vram_gb=18, recommended_vram_gb=24),
304294
}
305295

306296

@@ -325,12 +315,12 @@ def evaluate_model_status(
325315
max_vram = hw.get("max_vram_gb", 0)
326316
total_ram = hw.get("ram", {}).get("total_gb", 0)
327317

328-
if max_vram >= req["recommended_vram_gb"]:
318+
if max_vram >= req.recommended_vram_gb:
329319
return "recommended"
330-
if max_vram >= req["min_vram_gb"]:
320+
if max_vram >= req.min_vram_gb:
331321
return "capable"
332-
if total_ram >= req["min_ram_gb"] and max_vram == 0:
322+
if total_ram >= req.min_ram_gb and max_vram == 0:
333323
return "cpu_only"
334-
if total_ram >= req["min_ram_gb"]:
324+
if total_ram >= req.min_ram_gb:
335325
return "capable_slow"
336326
return "insufficient"

0 commit comments

Comments
 (0)