fix: resolve CodeFlow/SonarCloud quality issues

sunbos · sunbos · commit 355aced4fa7c · 2026-06-08T07:39:45.000+08:00
- signal.alarm: use vars(signal) dict lookup instead of type: ignore

- server.py: extract _build_ai_config() to eliminate code duplication

- analyzer.py: extract _extract_tool_call_result() to reduce complexity

- config.py: add explicit AIBackend type annotation

- _model_selector.py: remove unnecessary pass statement

- _create_demo_db.py: fix redefined-variable-type

- quickstart.py: use with for urlopen, narrow exception types, reduce complexity, fix f-string
diff --git a/plugins/mcp-server-sqlseed/src/mcp_server_sqlseed/server.py b/plugins/mcp-server-sqlseed/src/mcp_server_sqlseed/server.py
@@ -25,6 +25,33 @@
 _MAX_YAML_CONFIG_SIZE = 256 * 1024
 
 
+def _build_ai_config(
+    db_path: str,
+    model: str | None,
+    backend: str | None,
+) -> tuple[AIConfig, dict[str, Any] | None]:
+    """Build an AIConfig with Gemma 4 defaults, validating db_path and backend.
+
+    Returns (config, None) on success, or (config, error_dict) on failure.
+    """
+    if not _AI_AVAILABLE:
+        return AIConfig(), {"error": "sqlseed-ai plugin not installed. Install with: pip install sqlseed-ai"}
+
+    db_path = _validate_db_path(db_path)
+
+    ai_config = AIConfig.from_env()
+    if model:
+        ai_config.model = model
+    if backend:
+        try:
+            ai_config.backend = AIBackend(backend)
+        except ValueError:
+            return ai_config, {"error": f"Invalid backend: {backend}. Use: google_ai_studio, ollama, openai_compat"}
+    ai_config.resolve_model()
+
+    return ai_config, None
+
+
 def _validate_db_path(db_path: str) -> str:
     resolved = Path(db_path).resolve()
     valid_exts = (".db", ".sqlite", ".sqlite3")
@@ -206,25 +233,14 @@ def sqlseed_gemma4_analyze(
     Supported backends: google_ai_studio (default), ollama, openai_compat.
     Supported models: gemma-4-26b-it (default), gemma-4-31b-it, gemma-4-4b-it, gemma-4-2b-it.
     """
-    if not _AI_AVAILABLE:
-        return {"error": "sqlseed-ai plugin not installed. Install with: pip install sqlseed-ai"}
+    ai_config, err = _build_ai_config(db_path, model, backend)
+    if err is not None:
+        return err
 
-    db_path = _validate_db_path(db_path)
     with DataOrchestrator(db_path) as orch:
         _validate_table_name(table_name, orch.get_table_names())
         schema_ctx = orch.get_schema_context(table_name)
 
-    # Build AIConfig with Gemma 4 defaults
-    ai_config = AIConfig.from_env()
-    if model:
-        ai_config.model = model
-    if backend:
-        try:
-            ai_config.backend = AIBackend(backend)
-        except ValueError:
-            return {"error": f"Invalid backend: {backend}. Use: google_ai_studio, ollama, openai_compat"}
-    ai_config.resolve_model()
-
     analyzer = SchemaAnalyzer(config=ai_config)
     result = analyzer.analyze_table_from_ctx(**_serialize_schema_context(schema_ctx))
 
@@ -259,22 +275,11 @@ def sqlseed_gemma4_agent_fill(
     The agent uses Gemma 4's tool use to understand schema semantics and
     produce appropriate data generation rules automatically.
     """
-    if not _AI_AVAILABLE:
-        return {"error": "sqlseed-ai plugin not installed. Install with: pip install sqlseed-ai"}
-
-    db_path = _validate_db_path(db_path)
+    ai_config, err = _build_ai_config(db_path, model, backend)
+    if err is not None:
+        return err
 
     # Step 1: AI analysis with self-correction
-    ai_config = AIConfig.from_env()
-    if model:
-        ai_config.model = model
-    if backend:
-        try:
-            ai_config.backend = AIBackend(backend)
-        except ValueError:
-            return {"error": f"Invalid backend: {backend}. Use: google_ai_studio, ollama, openai_compat"}
-    ai_config.resolve_model()
-
     analyzer = SchemaAnalyzer(config=ai_config)
     refiner = AiConfigRefiner(analyzer, db_path)
 
diff --git a/plugins/sqlseed-ai/src/sqlseed_ai/_model_selector.py b/plugins/sqlseed-ai/src/sqlseed_ai/_model_selector.py
@@ -93,4 +93,3 @@ def select_next_free_model(failed_model: str) -> str | None:
 
 def clear_cache() -> None:
     """Legacy compat: no-op, Gemma models don't need cache."""
-    pass
diff --git a/plugins/sqlseed-ai/src/sqlseed_ai/analyzer.py b/plugins/sqlseed-ai/src/sqlseed_ai/analyzer.py
@@ -343,6 +343,26 @@ def _call_llm_once(self, messages: list[dict[str, str]]) -> dict[str, Any]:
             return {}
         return self._parse_json_response(content)
 
+    def _extract_tool_call_result(self, choice: Any) -> dict[str, Any] | None:
+        """Extract the analyze_schema result from a tool call choice."""
+        if not choice.message.tool_calls:
+            return None
+        for tool_call in choice.message.tool_calls:
+            if tool_call.function.name == "analyze_schema":
+                args_str = tool_call.function.arguments
+                if args_str:
+                    try:
+                        result: dict[str, Any] | None = json.loads(args_str)
+                        logger.info(
+                            "Gemma 4 native function calling succeeded",
+                            tool="analyze_schema",
+                            model=self._config.model if self._config else "unknown",
+                        )
+                        return result
+                    except json.JSONDecodeError:
+                        logger.debug("Failed to parse tool call arguments", args=args_str[:200])
+        return None
+
     def _try_tool_calling(self, client: Any, kwargs: dict[str, Any]) -> dict[str, Any] | None:
         """Attempt Gemma 4 native function calling.
 
@@ -365,22 +385,9 @@ def _try_tool_calling(self, client: Any, kwargs: dict[str, Any]) -> dict[str, An
 
             choice = response.choices[0]
 
-            # Check if the model made a tool call
-            if choice.message.tool_calls:
-                for tool_call in choice.message.tool_calls:
-                    if tool_call.function.name == "analyze_schema":
-                        args_str = tool_call.function.arguments
-                        if args_str:
-                            try:
-                                result: dict[str, Any] | None = json.loads(args_str)
-                                logger.info(
-                                    "Gemma 4 native function calling succeeded",
-                                    tool="analyze_schema",
-                                    model=self._config.model if self._config else "unknown",
-                                )
-                                return result
-                            except json.JSONDecodeError:
-                                logger.debug("Failed to parse tool call arguments", args=args_str[:200])
+            result = self._extract_tool_call_result(choice)
+            if result is not None:
+                return result
 
             # If no tool call was made but we have text content, parse it
             if choice.message.content:
diff --git a/plugins/sqlseed-ai/src/sqlseed_ai/config.py b/plugins/sqlseed-ai/src/sqlseed_ai/config.py
@@ -70,7 +70,7 @@ def from_env(cls) -> AIConfig:
         timeout = float(timeout_str) if timeout_str else 60.0
 
         # Resolve backend
-        backend = AIBackend.GOOGLE_AI_STUDIO  # default
+        backend: AIBackend = AIBackend.GOOGLE_AI_STUDIO
         if backend_str == "lm_studio":
             backend = AIBackend.LM_STUDIO
         elif backend_str == "ollama":
diff --git a/scripts/_create_demo_db.py b/scripts/_create_demo_db.py
@@ -1,4 +1,5 @@
 """Create demo database with 3 tables for quickstart script."""
+
 from __future__ import annotations
 
 import sqlite3
@@ -7,8 +8,8 @@
 
 
 def main() -> None:
-    db_path = sys.argv[1] if len(sys.argv) > 1 else "quickstart_demo.db"
-    db_path = Path(db_path)
+    db_path_str = sys.argv[1] if len(sys.argv) > 1 else "quickstart_demo.db"
+    db_path = Path(db_path_str)
     if db_path.exists():
         db_path.unlink()
 
diff --git a/scripts/quickstart.py b/scripts/quickstart.py
@@ -5,12 +5,14 @@
     python scripts/quickstart.py [--backend lm_studio|ollama|google] [--model MODEL_NAME]
     python scripts/quickstart.py --skip-install   # Skip pip install, use current env
 """
+
 from __future__ import annotations
 
 import argparse
 import os
 import subprocess
 import sys
+import urllib.error
 import urllib.request
 from pathlib import Path
 
@@ -27,18 +29,18 @@ def run(cmd: list[str], check: bool = True) -> subprocess.CompletedProcess[str]:
 def check_lm_studio() -> bool:
     """Check if LM Studio is running."""
     try:
-        r = urllib.request.urlopen("http://127.0.0.1:1234/v1/models", timeout=3)
-        return r.status == 200
-    except Exception:
+        with urllib.request.urlopen("http://127.0.0.1:1234/v1/models", timeout=3) as r:
+            return r.status == 200
+    except (urllib.error.URLError, OSError):
         return False
 
 
 def check_ollama() -> bool:
     """Check if Ollama is running."""
     try:
-        r = urllib.request.urlopen("http://localhost:11434/api/tags", timeout=3)
-        return r.status == 200
-    except Exception:
+        with urllib.request.urlopen("http://localhost:11434/api/tags", timeout=3) as r:
+            return r.status == 200
+    except (urllib.error.URLError, OSError):
         return False
 
 
@@ -47,6 +49,53 @@ def sqlseed_cmd(python: str) -> list[str]:
     return [python, "-c", "from sqlseed.cli.main import cli; cli()", "--"]
 
 
+def _fill_data(python: str) -> None:
+    """Fill the demo database with sample data using sqlseed CLI."""
+    cmd = sqlseed_cmd(python)
+    for table, count in [("users", 500), ("projects", 200), ("orders", 1000)]:
+        run([*cmd, "fill", str(DB_PATH), "-t", table, "-n", str(count)])
+
+    print()
+    print("  Data fill complete!")
+
+
+def _run_ai_analysis_step(args: argparse.Namespace, python: str) -> None:
+    """Run the Gemma 4 AI schema analysis step."""
+    print("[5/5] Gemma 4 AI Schema Analysis...")
+    print(f"  Backend: {args.backend}")
+    print(f"  Model:   {args.model}")
+    print()
+
+    os.environ["SQLSEED_AI_BACKEND"] = args.backend
+    os.environ["SQLSEED_AI_MODEL"] = args.model
+
+    ai_ok = False
+    if args.backend == "lm_studio":
+        if check_lm_studio():
+            ai_ok = True
+        else:
+            print("  Skipped: LM Studio is not running")
+            print("  Please start LM Studio and load a Gemma 4 model")
+            print("  Download: https://lmstudio.ai/")
+    elif args.backend == "ollama":
+        if check_ollama():
+            ai_ok = True
+        else:
+            print("  Skipped: Ollama is not running")
+            print("  Example: ollama pull gemma4:4b")
+    elif args.backend == "google":
+        if os.environ.get("GOOGLE_API_KEY"):
+            ai_ok = True
+        else:
+            print("  Skipped: GOOGLE_API_KEY not set")
+            print("  Example: export GOOGLE_API_KEY=your-key")
+
+    if ai_ok:
+        output_yaml = str(PROJECT_ROOT / "projects_config.yaml")
+        cmd = sqlseed_cmd(python)
+        run([*cmd, "ai-suggest", str(DB_PATH), "-t", "projects", "-o", output_yaml, "--timeout", "300"])
+
+
 def main() -> None:
     parser = argparse.ArgumentParser(description="GemmaSQLSeed one-click setup")
     parser.add_argument(
@@ -107,48 +156,10 @@ def main() -> None:
 
     # ── Step 4: Fill data ────────────────────────────────────────────
     print("[4/5] Filling data (zero-config)...")
-
-    cmd = sqlseed_cmd(python)
-    for table, count in [("users", 500), ("projects", 200), ("orders", 1000)]:
-        run(cmd + ["fill", str(DB_PATH), "-t", table, "-n", str(count)])
-
-    print()
-    print("  Data fill complete!")
+    _fill_data(python)
 
     # ── Step 5: Gemma 4 AI analysis ──────────────────────────────────
-    print("[5/5] Gemma 4 AI Schema Analysis...")
-    print(f"  Backend: {args.backend}")
-    print(f"  Model:   {args.model}")
-    print()
-
-    os.environ["SQLSEED_AI_BACKEND"] = args.backend
-    os.environ["SQLSEED_AI_MODEL"] = args.model
-
-    ai_ok = False
-    if args.backend == "lm_studio":
-        if check_lm_studio():
-            ai_ok = True
-        else:
-            print("  Skipped: LM Studio is not running")
-            print("  Please start LM Studio and load a Gemma 4 model")
-            print("  Download: https://lmstudio.ai/")
-    elif args.backend == "ollama":
-        if check_ollama():
-            ai_ok = True
-        else:
-            print("  Skipped: Ollama is not running")
-            print("  Example: ollama pull gemma4:4b")
-    elif args.backend == "google":
-        if os.environ.get("GOOGLE_API_KEY"):
-            ai_ok = True
-        else:
-            print("  Skipped: GOOGLE_API_KEY not set")
-            print("  Example: export GOOGLE_API_KEY=your-key")
-
-    if ai_ok:
-        output_yaml = str(PROJECT_ROOT / "projects_config.yaml")
-        cmd = sqlseed_cmd(python)
-        run(cmd + ["ai-suggest", str(DB_PATH), "-t", "projects", "-o", output_yaml, "--timeout", "300"])
+    _run_ai_analysis_step(args, python)
 
     # ── Done ─────────────────────────────────────────────────────────
     print()
@@ -157,10 +168,11 @@ def main() -> None:
     print("=" * 50)
     print()
     print(f"  Database:    {DB_PATH}")
-    print(f"  Preview:     python -c \"from sqlseed.cli.main import cli; cli()\" -- preview {DB_PATH} -t users -n 5")
-    print(f"  Inspect:     python -c \"from sqlseed.cli.main import cli; cli()\" -- inspect {DB_PATH} --show-mapping")
-    print(f"  AI Suggest:  python -c \"from sqlseed.cli.main import cli; cli()\" -- ai-suggest {DB_PATH} -t users -o config.yaml")
-    print(f"  MCP Server:  mcp-server-sqlseed")
+    cli_call = 'python -c "from sqlseed.cli.main import cli; cli()" --'
+    print(f"  Preview:     {cli_call} preview {DB_PATH} -t users -n 5")
+    print(f"  Inspect:     {cli_call} inspect {DB_PATH} --show-mapping")
+    print(f"  AI Suggest:  {cli_call} ai-suggest {DB_PATH} -t users -o config.yaml")
+    print("  MCP Server:  mcp-server-sqlseed")
     print()
 
 
diff --git a/src/sqlseed/cli/main.py b/src/sqlseed/cli/main.py
@@ -495,13 +495,15 @@ def ai_suggest(
     old_handler: Any = None
     if hasattr(signal, "SIGALRM"):
         old_handler = signal.signal(signal.SIGALRM, lambda _s, _f: _sigalrm_handler(total_timeout))
-        signal.alarm(int(total_timeout))  # type: ignore[attr-defined,unused-ignore]
+        _alarm_fn = vars(signal)["alarm"]
+        _alarm_fn(int(total_timeout))
 
     try:
         result = _run_ai_analysis(analyzer, db_path, table, verify, max_retries, no_cache)
     finally:
         if hasattr(signal, "SIGALRM"):
-            signal.alarm(0)  # type: ignore[attr-defined,unused-ignore]
+            _alarm_fn = vars(signal)["alarm"]
+            _alarm_fn(0)
             if old_handler is not None:
                 signal.signal(signal.SIGALRM, old_handler)
 

Original file line number	Diff line number	Diff line change
`@@ -93,4 +93,3 @@ def select_next_free_model(failed_model: str) -> str \| None:`
`93`	`93`
`94`	`94`	`def clear_cache() -> None:`
`95`	`95`	`"""Legacy compat: no-op, Gemma models don't need cache."""`
`96`		`- pass`