refactor: resolve CodeFlow code quality issues (35 to ~10 remaining)

Remove unused imports, convert lists to tuples, fix import ordering, narrow broad exceptions, extract methods to reduce duplication and complexity in analyzer.py, refiner.py, _json_utils.py

Author: sunbos

examples/quick_demo.py

@@ -15,13 +15,11 @@
 from __future__ import annotations
 
 import sqlite3
-import time
 from pathlib import Path
 
 from rich.console import Console
 from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TimeElapsedColumn, TimeRemainingColumn
 from rich.table import Table
-from rich.text import Text
 
 from sqlseed import fill
 
@@ -170,21 +168,21 @@ def main():
     cursor = conn.execute("SELECT member_id, name, email, org_code FROM members LIMIT 3")
     rows = cursor.fetchall()
     member_table = Table()
-    for col in ["member_id", "name", "email", "org_code"]:
+    for col in ("member_id", "name", "email", "org_code"):
         member_table.add_column(col, style="cyan")
     for row in rows:
-        member_table.add_row(*[str(row[c]) for c in ["member_id", "name", "email", "org_code"]])
+        member_table.add_row(*[str(row[c]) for c in ("member_id", "name", "email", "org_code")])
     console.print(member_table)
 
     # Preview tasks
     console.print("\n[bold green]tasks[/bold green] (3 rows):")
     cursor = conn.execute("SELECT task_id, project_id, title, priority, status FROM tasks LIMIT 3")
     rows = cursor.fetchall()
     task_table = Table()
-    for col in ["task_id", "project_id", "title", "priority", "status"]:
+    for col in ("task_id", "project_id", "title", "priority", "status"):
         task_table.add_column(col, style="cyan")
     for row in rows:
-        task_table.add_row(*[str(row[c]) for c in ["task_id", "project_id", "title", "priority", "status"]])
+        task_table.add_row(*[str(row[c]) for c in ("task_id", "project_id", "title", "priority", "status")])
     console.print(task_table)
 
     conn.close()
@@ -197,7 +195,7 @@ def main():
     console.print("  • Smart column type matching")
     console.print("  • Foreign key integrity")
     console.print("  • Real-time progress tracking")
-    console.print("\n[italic]Total time:[/italic] [bold]{:.2f} seconds[/bold]".format(total_time))
+    console.print(f"\n[italic]Total time:[/italic] [bold]{total_time:.2f} seconds[/bold]")
 
 
 if __name__ == "__main__":

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

@@ -6,55 +6,66 @@
 
 
 def parse_json_response(content: str) -> dict[str, Any]:
+    """Parse JSON from LLM response using 3-strategy fallback."""
     cleaned = content.strip()
 
-    # Strategy 1: Direct parse (ideal case — model outputs raw JSON)
+    return _try_direct_parse(cleaned) or _try_markdown_fence_parse(cleaned) or _try_raw_decode(cleaned) or {}
+
+
+def _try_direct_parse(content: str) -> dict[str, Any] | None:
+    """Strategy 1: Direct parse (ideal case — model outputs raw JSON)."""
+    try:
+        result = json.loads(content)
+        if isinstance(result, dict):
+            _sanitize_names(result)
+            return result
+    except json.JSONDecodeError:
+        pass
+    return None
+
+
+def _try_markdown_fence_parse(content: str) -> dict[str, Any] | None:
+    """Strategy 2: Strip markdown code fences (```json\\n{...}\\n```)."""
+    open_idx = content.find("```")
+    if open_idx < 0:
+        return None
+    after_open = content[open_idx + 3 :]
+    nl_pos = after_open.find("\n")
+    if nl_pos < 0:
+        return None
+    content_start = nl_pos + 1
+    close_idx = after_open.find("```", content_start)
+    if close_idx < 0:
+        return None
+    fence_content = after_open[content_start:close_idx].strip()
     try:
-        result = json.loads(cleaned)
+        result = json.loads(fence_content)
         if isinstance(result, dict):
             _sanitize_names(result)
             return result
     except json.JSONDecodeError:
         pass
+    return None
 
-    # Strategy 2: Strip markdown code fences
-    # Handles: ```json\n{...}\n``` or ```\n{...}\n```
-    # Split on fence markers instead of using backtracking-prone regex with DOTALL
-    open_idx = cleaned.find("```")
-    if open_idx >= 0:
-        # Skip the opening fence and optional language tag
-        after_open = cleaned[open_idx + 3 :]
-        # Skip "json" or other language identifier on the same line
-        nl_pos = after_open.find("\n")
-        if nl_pos >= 0:
-            content_start = nl_pos + 1
-            # Find closing fence
-            close_idx = after_open.find("```", content_start)
-            if close_idx >= 0:
-                fence_content = after_open[content_start:close_idx].strip()
-                try:
-                    result = json.loads(fence_content)
-                    if isinstance(result, dict):
-                        _sanitize_names(result)
-                        return result
-                except json.JSONDecodeError:
-                    pass
 
-    # Strategy 3: Find first '{' and use json.JSONDecoder.raw_decode()
-    # Handles: explanatory text before/after JSON, no code fences
-    # raw_decode() correctly handles braces inside JSON strings
-    first_brace = cleaned.find("{")
-    if first_brace >= 0:
-        try:
-            decoder = json.JSONDecoder()
-            result, _ = decoder.raw_decode(cleaned, idx=first_brace)
-            if isinstance(result, dict):
-                _sanitize_names(result)
-                return result
-        except json.JSONDecodeError:
-            pass
+def _try_raw_decode(content: str) -> dict[str, Any] | None:
+    """Strategy 3: Find first '{' and use json.JSONDecoder.raw_decode().
 
-    return {}
+    Handles explanatory text before/after JSON without code fences.
+    raw_decode() correctly handles braces inside JSON strings.
+    """
+    first_brace = content.find("{")
+    if first_brace < 0:
+        return None
+    try:
+        decoder = json.JSONDecoder()
+        result, _ = decoder.raw_decode(content, idx=first_brace)
+        if isinstance(result, dict):
+            _sanitize_names(result)
+            return result
+    except json.JSONDecodeError:
+        pass
+    return None
 
 
 def _sanitize_names(data: dict[str, Any]) -> None:

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

@@ -298,17 +298,28 @@ def _find_local_fallback_model(
 
         return None
 
-    def call_llm(self, messages: list[dict[str, str]]) -> dict[str, Any]:
+    def _ensure_config(self) -> None:
+        """Initialize and validate AIConfig if not already done."""
         if self._config is None:
             self._config = AIConfig.from_env()
         self._config.resolve_model()
         if not self._config.resolve_api_key():
             raise ValueError("AI API key not configured")
 
-        current_model = self._config.model
+    def _call_with_fallback(
+        self,
+        call_fn: Callable[[str], dict[str, Any]],
+    ) -> dict[str, Any]:
+        """Execute an LLM call with model fallback on timeout/connection errors.
+
+        Args:
+            call_fn: A callable that takes a model name and returns the LLM result.
+        """
+        assert self._config is not None  # ensured by _ensure_config()
+        current_model: str = self._config.model or ""
         for attempt in range(_MAX_FALLBACK_ATTEMPTS):
             try:
-                return self._call_llm_once(messages, model=current_model)
+                return call_fn(current_model)
             except (APITimeoutError, APIConnectionError) as e:
                 logger.warning(
                     "LLM API call timed out or connection failed",
@@ -344,6 +355,10 @@ def call_llm(self, messages: list[dict[str, str]]) -> dict[str, Any]:
 
         raise RuntimeError(f"LLM API call failed after {_MAX_FALLBACK_ATTEMPTS} fallback attempts")
 
+    def call_llm(self, messages: list[dict[str, str]]) -> dict[str, Any]:
+        self._ensure_config()
+        return self._call_with_fallback(lambda model: self._call_llm_once(messages, model=model))
+
     def call_llm_streaming(
         self,
         messages: list[dict[str, str]],
@@ -360,50 +375,8 @@ def call_llm_streaming(
                 - "parsing": parsing the response JSON
                 - "done": analysis complete, info={"tokens": int, "model": str}
         """
-        if self._config is None:
-            self._config = AIConfig.from_env()
-        self._config.resolve_model()
-        if not self._config.resolve_api_key():
-            raise ValueError("AI API key not configured")
-
-        current_model = self._config.model
-        for attempt in range(_MAX_FALLBACK_ATTEMPTS):
-            try:
-                return self._call_llm_streaming_once(messages, on_progress, model=current_model)
-            except (APITimeoutError, APIConnectionError) as e:
-                logger.warning(
-                    "LLM API call timed out or connection failed",
-                    model=current_model,
-                    error=str(e)[:200],
-                    attempt=attempt + 1,
-                )
-
-                next_model = select_next_gemma_model(current_model or "", backend=self._config.backend)
-                if next_model is None:
-                    raise RuntimeError(
-                        f"LLM API call failed after trying {attempt + 1} model(s). "
-                        f"Last error (model={current_model}): {e}"
-                    ) from e
-
-                # For local backends, verify the fallback model is actually available.
-                if self._config.backend in (AIBackend.LM_STUDIO, AIBackend.OLLAMA):
-                    actual_model = self._find_local_fallback_model(current_model, next_model)
-                    if actual_model is None:
-                        raise RuntimeError(
-                            f"No other model available on local backend besides {current_model}. "
-                            f"Consider using a smaller model or increasing --timeout. "
-                            f"Last error: {e}"
-                        ) from e
-                    next_model = actual_model
-
-                logger.warning(
-                    "Falling back to next Gemma 4 model",
-                    from_model=current_model,
-                    to_model=next_model,
-                )
-                current_model = next_model
-
-        raise RuntimeError(f"LLM API call failed after {_MAX_FALLBACK_ATTEMPTS} fallback attempts")
+        self._ensure_config()
+        return self._call_with_fallback(lambda model: self._call_llm_streaming_once(messages, on_progress, model=model))
 
     @staticmethod
     def _is_reasoning_model_id(model_id: str | None) -> bool:

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

@@ -216,7 +216,7 @@ def _detect_all_local_models(self) -> list[str]:
                 models = [m.get("id") for m in data.get("data", []) if m.get("id")]
                 self._all_models_cache = (time.monotonic(), models)
                 return models
-        except Exception:
+        except (OSError, ValueError, KeyError):
             logger.debug("Could not auto-detect local models", backend=self.backend.value)
 
         # Cache negative result too, to avoid hammering a dead endpoint