refactor(ai): eliminate CodeFlow v10 6 code-duplication warnings via best practices

1. refiner.py: Add AiConfigRefiner.from_config() classmethod; change AISuggestionFailedError to inherit from RuntimeError (fixes bug where ai_commands.py did not catch it). 2. __init__.py: Re-export AIBackend, GemmaModel, AiConfigRefiner, AISuggestionFailedError + add __all__. 3. ai_commands.py + mcp.py: Use package-level imports. 4. mcp.py: Use from_config() at 2 call sites. 5. test files: Extract shared import block to tests/_ai_imports.py + helper in test_stream.py. Validation: ruff/mypy pass, 1594 pytest passed.

Author: sunbos

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

@@ -27,13 +27,24 @@
 
 from sqlseed_ai.ai_mediator import apply_ai_suggestions
 from sqlseed_ai.analyzer import SchemaAnalyzer
-from sqlseed_ai.config import AIConfig
+from sqlseed_ai.config import AIBackend, AIConfig, GemmaModel
+from sqlseed_ai.refiner import AiConfigRefiner, AISuggestionFailedError
 
 from sqlseed._utils.logger import get_logger
 from sqlseed.plugins.hookspecs import hookimpl
 
 logger = get_logger(__name__)
 
+__all__ = [
+    "AIBackend",
+    "AIConfig",
+    "AISuggestionFailedError",
+    "AiConfigRefiner",
+    "GemmaModel",
+    "SchemaAnalyzer",
+    "apply_ai_suggestions",
+]
+
 _SIMPLE_COL_RE = re.compile(
     r"(^|[_\s])("
     r"name|email|phone|address|url|uuid|"

plugins/sqlseed-ai/src/sqlseed_ai/cli/ai_commands.py

@@ -19,9 +19,7 @@
 from rich.console import Console
 from rich.live import Live
 from rich.text import Text
-from sqlseed_ai.analyzer import SchemaAnalyzer
-from sqlseed_ai.config import AIBackend, AIConfig
-from sqlseed_ai.refiner import AiConfigRefiner
+from sqlseed_ai import AIBackend, AIConfig, AiConfigRefiner, SchemaAnalyzer
 
 # sanitize_table_config lives in the sqlseed-cli package; this is the only
 # cross-plugin import permitted per ARCHITECTURE.md Section 4 (sqlseed-ai

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

@@ -29,10 +29,8 @@
 except ImportError as _exc:  # pragma: no cover - import error path
     raise ImportError("mcp SDK not installed. Install with: pip install 'sqlseed-ai[mcp]'") from _exc
 
+from sqlseed_ai import AIBackend, AIConfig, AiConfigRefiner, AISuggestionFailedError, GemmaModel, SchemaAnalyzer
 from sqlseed_ai._hardware import MODEL_REQUIREMENTS, detect_hardware, evaluate_model_status
-from sqlseed_ai.analyzer import SchemaAnalyzer
-from sqlseed_ai.config import AIBackend, AIConfig, GemmaModel
-from sqlseed_ai.refiner import AiConfigRefiner, AISuggestionFailedError
 
 from sqlseed._utils.logger import get_logger
 from sqlseed._utils.paths import validate_db_target as _validate_db_target
@@ -114,8 +112,7 @@ def sqlseed_ai_generate_yaml(
         )
         ai_config.model = ai_config.resolve_model()
 
-        analyzer = SchemaAnalyzer(config=ai_config)
-        refiner = AiConfigRefiner(analyzer, db_path)
+        refiner = AiConfigRefiner.from_config(ai_config, db_path)
 
         result = refiner.generate_and_refine(
             table_name=table_name,
@@ -201,8 +198,7 @@ def sqlseed_gemma4_agent_fill(
         ai_config = _build_ai_config(db_path, model, backend)
 
         # Step 1: AI analysis with self-correction
-        analyzer = SchemaAnalyzer(config=ai_config)
-        refiner = AiConfigRefiner(analyzer, db_path)
+        refiner = AiConfigRefiner.from_config(ai_config, db_path)
 
         try:
             ai_result = refiner.generate_and_refine(

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

@@ -31,6 +31,7 @@
     from collections.abc import Callable
 
     from sqlseed_ai.analyzer import SchemaAnalyzer
+    from sqlseed_ai.config import AIConfig
 
 logger = get_logger(__name__)
 
@@ -47,8 +48,14 @@ def __init__(self) -> None:
         self.min_prompt_level: int = 0
 
 
-class AISuggestionFailedError(Exception):
-    """Raised when AI config generation/refinement cannot produce a valid config."""
+class AISuggestionFailedError(RuntimeError):
+    """Raised when AI config generation/refinement cannot produce a valid config.
+
+    Inherits from :class:`RuntimeError` so that callers catching
+    ``(ValueError, RuntimeError, OSError)`` (the standard recoverable-error
+    tuple used across sqlseed-ai) also catch this exception without needing
+    to import it explicitly.
+    """
 
 
 class AiConfigRefiner:
@@ -79,6 +86,34 @@ def __init__(
         self._db_path = db_path
         self._cache_dir = Path(cache_dir) if cache_dir else get_cache_dir("ai_configs")
 
+    @classmethod
+    def from_config(
+        cls,
+        ai_config: AIConfig,
+        db_path: str,
+        *,
+        cache_dir: str | None = None,
+    ) -> AiConfigRefiner:
+        """Create a refiner with an internally-constructed analyzer.
+
+        Convenience factory for callers that don't need the
+        :class:`SchemaAnalyzer` separately (e.g., MCP tools that only call
+        ``generate_and_refine``). Callers that need the analyzer for other
+        operations (e.g., CLI streaming display) should construct the
+        analyzer explicitly and use the regular constructor.
+
+        Args:
+            ai_config: The AI configuration to build the analyzer from.
+            db_path: Path to the database file (or URL) to validate against.
+            cache_dir: Optional override for the cache directory.
+
+        Returns:
+            A new :class:`AiConfigRefiner` instance.
+        """
+        from sqlseed_ai.analyzer import SchemaAnalyzer
+
+        return cls(SchemaAnalyzer(config=ai_config), db_path, cache_dir=cache_dir)
+
     def _handle_generation_failure(self, error: ErrorSummary, attempt: int, max_retries: int) -> None:
         """Decide whether to retry or raise after an LLM generation failure.
 

plugins/sqlseed-ai/tests/test_ai_analyzer_streaming.py

@@ -15,16 +15,20 @@
 
 import pytest
 
-# Optional sqlseed-ai / openai imports. The try/except block is the standard
-# pytest pattern for optional test deps — pylint exempts imports inside
-# try/except from wrong-import-position/wrong-import-order, so this block
-# does not trigger those warnings (unlike the ensure_ai_deps() call pattern
-# which placed imports after a function call and triggered 8 warnings).
+# Re-export optional AI deps from a shared module to avoid cross-file import
+# block duplication. The try/except + pytest.skip(allow_module_level=True)
+# pattern is required because pytest.skip only takes effect in the module
+# where it is called.
 try:
-    from openai import APIConnectionError, APIError, APITimeoutError
-    from sqlseed_ai.analyzer import SchemaAnalyzer
-    from sqlseed_ai.config import AIBackend, AIConfig
-    from sqlseed_ai.exceptions import ContextOverflowError
+    from tests._ai_imports import (
+        AIBackend,
+        AIConfig,
+        APIConnectionError,
+        APIError,
+        APITimeoutError,
+        ContextOverflowError,
+        SchemaAnalyzer,
+    )
 except ImportError:
     pytest.skip("sqlseed-ai plugin not installed", allow_module_level=True)
 

plugins/sqlseed-ai/tests/test_ai_caller.py

@@ -17,16 +17,20 @@
 
 import pytest
 
-# Optional sqlseed-ai / openai imports. The try/except block is the standard
-# pytest pattern for optional test deps — pylint exempts imports inside
-# try/except from wrong-import-position/wrong-import-order, so this block
-# does not trigger those warnings (unlike the ensure_ai_deps() call pattern
-# which placed imports after a function call and triggered 8 warnings).
+# Re-export optional AI deps from a shared module to avoid cross-file import
+# block duplication. The try/except + pytest.skip(allow_module_level=True)
+# pattern is required because pytest.skip only takes effect in the module
+# where it is called.
 try:
-    from openai import APIConnectionError, APIError, APITimeoutError
-    from sqlseed_ai.analyzer import SchemaAnalyzer
-    from sqlseed_ai.config import AIBackend, AIConfig
-    from sqlseed_ai.exceptions import ContextOverflowError
+    from tests._ai_imports import (
+        AIBackend,
+        AIConfig,
+        APIConnectionError,
+        APIError,
+        APITimeoutError,
+        ContextOverflowError,
+        SchemaAnalyzer,
+    )
 except ImportError:
     pytest.skip("sqlseed-ai plugin not installed", allow_module_level=True)
 

tests/_ai_imports.py

@@ -0,0 +1,28 @@
+"""Shared AI test imports (re-exported for test modules).
+
+Centralizes the optional ``sqlseed_ai`` / ``openai`` imports so that test
+modules can import them via ``from tests._ai_imports import ...`` inside a
+try/except block, avoiding cross-file import block duplication.
+
+Test modules still own the try/except + ``pytest.skip(allow_module_level=True)``
+logic because ``pytest.skip`` only takes effect in the module where it is
+called — this module deliberately has no skip logic so the caller controls
+module-level skip behavior.
+"""
+
+from __future__ import annotations
+
+from openai import APIConnectionError, APIError, APITimeoutError
+from sqlseed_ai.analyzer import SchemaAnalyzer
+from sqlseed_ai.config import AIBackend, AIConfig
+from sqlseed_ai.exceptions import ContextOverflowError
+
+__all__ = [
+    "AIBackend",
+    "AIConfig",
+    "APIConnectionError",
+    "APIError",
+    "APITimeoutError",
+    "ContextOverflowError",
+    "SchemaAnalyzer",
+]

tests/test_core/test_stream.py

@@ -497,6 +497,24 @@ def test_foreign_key_fallback_without_ref_values(self) -> None:
         assert second_call.kwargs == {"min_value": 1, "max_value": 50}
 
 
+def _make_provider_with_missing_native_attr(
+    provider_name: str,
+    native_attr: str,
+    del_attr: str,
+) -> MagicMock:
+    """Build a mock provider whose native object has an attribute deleted.
+
+    Used to verify ``_try_native_method`` returns ``_NATIVE_MISS`` when the
+    native target (faker method or mimesis path component) is missing.
+    """
+    native_obj = MagicMock()
+    delattr(native_obj, del_attr)
+    provider = MagicMock()
+    provider.name = provider_name
+    setattr(provider, native_attr, native_obj)
+    return provider
+
+
 class TestTryNativeMethod:
     """Tests for _try_native_method, _try_faker_native, _try_mimesis_native."""
 
@@ -533,12 +551,7 @@ def test_try_faker_native_calls_method_when_available(self) -> None:
         fake_faker.email.assert_called_once_with(domain="example.com")
 
     def test_try_faker_native_returns_miss_when_method_not_found(self) -> None:
-        fake_faker = MagicMock()
-        # Configure so getattr returns None for the missing method
-        del fake_faker.nonexistent_method
-        provider = MagicMock()
-        provider.name = "faker"
-        provider._faker = fake_faker
+        provider = _make_provider_with_missing_native_attr("faker", "_faker", "nonexistent_method")
 
         spec = GeneratorSpec(generator_name="string", native_faker_method="nonexistent_method")
         stream = make_stream([], provider)
@@ -598,12 +611,7 @@ def test_try_mimesis_native_walks_dotted_path(self) -> None:
         generic.person.full_name.assert_called_once_with()
 
     def test_try_mimesis_native_returns_miss_on_invalid_path(self) -> None:
-        # Path component doesn't exist → returns _NATIVE_MISS
-        generic = MagicMock()
-        del generic.nonexistent
-        provider = MagicMock()
-        provider.name = "mimesis"
-        provider._generic = generic
+        provider = _make_provider_with_missing_native_attr("mimesis", "_generic", "nonexistent")
 
         spec = GeneratorSpec(
             generator_name="string",