refactor(devgen/providers): shared provider base class

Abstract out generate validation and error handling into BaseProvider, eliminating duplicated logic across Anthropic, OpenRouter, and OpenAI providers.

devgen/providers/anthropic.py

@@ -1,33 +1,19 @@
 import anthropic
 
-from devgen.utils import format_token_limit_error, is_token_limit_error
+from devgen.providers.base import BaseProvider
 
 
-class AnthropicProvider:
+class AnthropicProvider(BaseProvider):
     """Generates content using Anthropic's Claude models."""
 
-    def generate(
-        self, prompt: str, api_key: str, model: str = "claude-3-opus-20240229", **kwargs
-    ) -> str:
-        """Generates a response using the Anthropic API."""
-        if not api_key:
-            raise ValueError(
-                "Anthropic API key is missing. "
-                "Set it via `devgen setup config` or pass --api-key."
-            )
+    DISPLAY_NAME = "Anthropic"
+    DEFAULT_MODEL = "claude-3-opus-20240229"
 
-        try:
-            client = anthropic.Anthropic(api_key=api_key)
-            message = client.messages.create(
-                model=model,
-                max_tokens=1024,
-                messages=[{"role": "user", "content": prompt}],
-            )
-            return message.content[0].text.strip()
-        except Exception as e:
-            if is_token_limit_error(e):
-                raise RuntimeError(format_token_limit_error("Anthropic", e)) from e
-            raise RuntimeError(
-                f"Anthropic generation failed: {e}. "
-                "Check the model id and your account credits."
-            ) from e
+    def _generate(self, prompt, api_key, model, **kwargs):
+        client = anthropic.Anthropic(api_key=api_key)
+        message = client.messages.create(
+            model=model,
+            max_tokens=1024,
+            messages=[{"role": "user", "content": prompt}],
+        )
+        return message.content[0].text

devgen/providers/base.py

@@ -0,0 +1,90 @@
+"""Base class for AI providers.
+
+Every provider implements :meth:`_generate` and inherits the standard
+contract: API-key validation, token-limit detection, and friendly
+RuntimeError wrapping. Providers with provider-specific error shapes
+(eg. Hugging Face's HTTPError / RequestException split) override
+:meth:`_handle_error`.
+"""
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+from devgen.utils import format_token_limit_error, is_token_limit_error
+
+
+class BaseProvider(ABC):
+    """Abstract base for all AI providers."""
+
+    #: Human-readable name used in error messages.
+    DISPLAY_NAME: str = ""
+    #: Whether the provider requires an API key.
+    REQUIRES_API_KEY: bool = True
+    #: Default model id when the user doesn't specify one.
+    DEFAULT_MODEL: str = ""
+
+    def generate(
+        self,
+        prompt: str,
+        api_key: str | None = None,
+        model: str | None = None,
+        **kwargs: Any,
+    ) -> str:
+        """Public entry point. Validates input, dispatches, and wraps errors.
+
+        Args:
+            prompt: The prompt to send to the model.
+            api_key: Provider API key. May be ``None`` for local providers
+                (see :attr:`REQUIRES_API_KEY`).
+            model: Model id. Falls back to :attr:`DEFAULT_MODEL` when ``None``.
+            **kwargs: Forwarded to :meth:`_generate`.
+
+        Returns:
+            The generated text, stripped.
+
+        Raises:
+            ValueError: If the provider requires an API key and none was given.
+            RuntimeError: For any underlying failure, with a friendly message.
+        """
+        if self.REQUIRES_API_KEY and not api_key:
+            raise ValueError(
+                f"{self.DISPLAY_NAME} API key is missing. "
+                "Set it via `devgen setup config` or pass --api-key."
+            )
+        chosen_model = model or self.DEFAULT_MODEL
+        try:
+            return self._generate(prompt, api_key, chosen_model, **kwargs).strip()
+        except Exception as e:
+            self._handle_error(e)
+
+    def _handle_error(self, error: Exception) -> None:
+        """Translate an internal exception into a user-facing RuntimeError.
+
+        Token-limit / context-window errors are reformatted with
+        :func:`format_token_limit_error`; everything else is wrapped in a
+        generic provider message. Subclasses override this to add
+        provider-specific error shapes (eg. HTTPError vs RequestException).
+        """
+        if is_token_limit_error(error):
+            raise RuntimeError(
+                format_token_limit_error(self.DISPLAY_NAME, error)
+            ) from error
+        raise RuntimeError(
+            f"{self.DISPLAY_NAME} request failed: {error}. "
+            "Check the model name, API key permissions, and your account quota."
+        ) from error
+
+    @abstractmethod
+    def _generate(
+        self,
+        prompt: str,
+        api_key: str | None,
+        model: str,
+        **kwargs: Any,
+    ) -> str:
+        """Provider-specific generation. Return the raw generated text.
+
+        Implementations may raise any exception; :meth:`_handle_error` will
+        translate it. Re-raise an exception to defer to default handling.
+        """
+        raise NotImplementedError

devgen/providers/gemini.py

@@ -1,31 +1,22 @@
 import time
+
 from google import genai
 from google.genai import types
 
-from devgen.utils import format_token_limit_error, is_token_limit_error
+from devgen.providers.base import BaseProvider
+from devgen.utils import is_token_limit_error
 
 
-class GeminiProvider:
+class GeminiProvider(BaseProvider):
     """Generates content using Google's Gemini models."""
 
-    def generate(
-        self, prompt: str, api_key: str, model: str = "gemini-2.5-flash", **kwargs
-    ) -> str:
-        """Generates a response using the Gemini API.
-
-        Default model is updated to gemini-2.5-flash which is the latest
-        recommended model.
-        """
-        if not api_key:
-            raise ValueError(
-                "Gemini API key is missing. "
-                "Set it via `devgen setup config` or pass --api-key."
-            )
-
-        client = genai.Client(api_key=api_key)
+    DISPLAY_NAME = "Gemini"
+    DEFAULT_MODEL = "gemini-2.5-flash"
 
+    def _generate(self, prompt, api_key, model, **kwargs):
         max_retries = kwargs.get("max_retries", 5)
-        base_delay = kwargs.get("retry_delay", 10)  # 10 seconds base delay for 429
+        base_delay = kwargs.get("retry_delay", 10)  # 10s base for 429 backoff
+        client = genai.Client(api_key=api_key)
 
         for attempt in range(max_retries + 1):
             try:
@@ -39,38 +30,34 @@ class GeminiProvider:
                         max_output_tokens=kwargs.get("max_output_tokens", 2048),
                     ),
                 )
-                return response.text.strip()
+                return response.text
             except Exception as e:
-                error_msg = str(e).upper()
-                # Token / context-window overflow is not retryable — no point
-                # backing off, the prompt is just too big.
                 if is_token_limit_error(e):
-                    raise RuntimeError(format_token_limit_error("Gemini", e)) from e
-                # Check for 429 or ResourceExhausted in common error formats
-                if any(
-                    x in error_msg
-                    for x in [
-                        "429",
-                        "RESOURCE_EXHAUSTED",
-                        "QUOTA_EXCEEDED",
-                        "THROTTLED",
-                    ]
-                ):
-                    if attempt < max_retries:
-                        delay = base_delay * (2**attempt)
-                        print(
-                            f"\n[Gemini] Quota limit hit (429). This is often a temporary rate limit (RPM). "
-                            f"Waiting {delay}s and retrying... (Attempt {attempt + 1}/{max_retries})"
-                        )
-                        time.sleep(delay)
-                        continue
-                    else:
-                        raise RuntimeError(
-                            "Gemini API quota exhausted (429) after multiple retries.\n\n"
-                            "POSSIBLE CAUSES:\n"
-                            "1. Rate Limit (RPM/RPD): You are sending requests too fast for the Free Tier.\n"
-                            "2. Daily Limit: You have reached the maximum free requests allowed for today (typically 1,500).\n"
-                            "3. Credit/Billing: If on a paid plan, your credit may be exhausted.\n\n"
-                            "Please check your usage metrics at: https://aistudio.google.com/app/usage"
-                        ) from e
-                raise RuntimeError(f"Gemini generation failed: {e}") from e
+                    raise  # base class handles with friendly message
+                if self._is_quota_error(e) and attempt < max_retries:
+                    delay = base_delay * (2**attempt)
+                    print(
+                        f"\n[Gemini] Quota limit hit (429). This is often a temporary "
+                        f"rate limit (RPM). Waiting {delay}s and retrying... "
+                        f"(Attempt {attempt + 1}/{max_retries})"
+                    )
+                    time.sleep(delay)
+                    continue
+                if self._is_quota_error(e):
+                    raise RuntimeError(
+                        "Gemini API quota exhausted (429) after multiple retries.\n\n"
+                        "POSSIBLE CAUSES:\n"
+                        "1. Rate Limit (RPM/RPD): You are sending requests too fast for the Free Tier.\n"
+                        "2. Daily Limit: You have reached the maximum free requests allowed for today (typically 1,500).\n"
+                        "3. Credit/Billing: If on a paid plan, your credit may be exhausted.\n\n"
+                        "Please check your usage metrics at: https://aistudio.google.com/app/usage"
+                    ) from e
+                raise
+
+    @staticmethod
+    def _is_quota_error(error: Exception) -> bool:
+        text = str(error).upper()
+        return any(
+            marker in text
+            for marker in ("429", "RESOURCE_EXHAUSTED", "QUOTA_EXCEEDED", "THROTTLED")
+        )

devgen/providers/huggingface.py

@@ -1,34 +1,19 @@
 import requests
 
+from devgen.providers.base import BaseProvider
 from devgen.utils import format_token_limit_error, is_token_limit_error
 
 
-class HuggingfaceProvider:
+class HuggingfaceProvider(BaseProvider):
     """Generates content using Hugging Face Inference API."""
 
+    DISPLAY_NAME = "Hugging Face"
     API_URL_TEMPLATE = "https://api-inference.huggingface.co/models/{model}"
+    DEFAULT_MODEL = "mistralai/Mistral-7B-Instruct-v0.2"
 
-    def generate(
-        self,
-        prompt: str,
-        api_key: str,
-        model: str = "mistralai/Mistral-7B-Instruct-v0.2",
-        **kwargs,
-    ) -> str:
-        """Generates a response using Hugging Face API."""
-        if not api_key:
-            raise ValueError(
-                "Hugging Face API token is missing. "
-                "Set it via `devgen setup config` or pass --api-key."
-            )
-
+    def _generate(self, prompt, api_key, model, **kwargs):
         api_url = self.API_URL_TEMPLATE.format(model=model)
         headers = {"Authorization": f"Bearer {api_key}"}
-
-        # HF models often expect specific prompting formats, but we'll send raw prompt
-        # Some models are text-generation, some are conversational.
-        # Assuming text-generation for generic usage.
-
         payload = {
             "inputs": prompt,
             "parameters": {"max_new_tokens": 500, "return_full_text": False},
@@ -38,31 +23,12 @@ class HuggingfaceProvider:
             response = requests.post(api_url, headers=headers, json=payload)
             response.raise_for_status()
             result = response.json()
-
-            if isinstance(result, list) and result and "generated_text" in result[0]:
-                return result[0]["generated_text"].strip()
-            elif isinstance(result, dict) and "error" in result:
-                err = result["error"]
-                if "is currently loading" in str(err).lower():
-                    raise RuntimeError(
-                        f"Hugging Face model is loading: {err}. "
-                        "Wait a minute and retry, or pick a different model."
-                    )
-                if is_token_limit_error(err):
-                    raise RuntimeError(
-                        format_token_limit_error("Hugging Face", err)
-                    ) from err
-                raise RuntimeError(f"Hugging Face API error: {err}")
-            else:
-                raise RuntimeError(
-                    "Hugging Face returned an unexpected response shape. "
-                    "The model may not support text-generation."
-                )
-
         except requests.HTTPError as e:
-            status = e.response.status_code if e.response is not None else "?"
             if is_token_limit_error(e):
-                raise RuntimeError(format_token_limit_error("Hugging Face", e)) from e
+                raise RuntimeError(
+                    format_token_limit_error(self.DISPLAY_NAME, e)
+                ) from e
+            status = e.response.status_code if e.response is not None else "?"
             raise RuntimeError(
                 f"Hugging Face request failed (HTTP {status}): {e}. "
                 "Check the model id and your token permissions."
@@ -71,9 +37,18 @@ class HuggingfaceProvider:
             raise RuntimeError(
                 f"Hugging Face network error: {e}. Check your connection."
             ) from e
-        except RuntimeError:
-            raise
-        except Exception as e:
-            if is_token_limit_error(e):
-                raise RuntimeError(format_token_limit_error("Hugging Face", e)) from e
-            raise RuntimeError(f"Hugging Face generation failed: {e}") from e
+
+        if isinstance(result, list) and result and "generated_text" in result[0]:
+            return result[0]["generated_text"]
+        if isinstance(result, dict) and "error" in result:
+            err = result["error"]
+            if "is currently loading" in str(err).lower():
+                raise RuntimeError(
+                    f"Hugging Face model is loading: {err}. "
+                    "Wait a minute and retry, or pick a different model."
+                )
+            raise RuntimeError(f"Hugging Face API error: {err}")
+        raise RuntimeError(
+            "Hugging Face returned an unexpected response shape. "
+            "The model may not support text-generation."
+        )

devgen/providers/ollama.py

@@ -1,54 +1,51 @@
+from openai import OpenAI
+
+from devgen.providers.base import BaseProvider
 from devgen.utils import format_token_limit_error, is_token_limit_error
 
 
-class OllamaProvider:
+class OllamaProvider(BaseProvider):
     """Generates content using a local Ollama server.
 
     Ollama exposes an OpenAI-compatible endpoint at /v1/chat/completions,
-    so we can reuse the openai client. The host defaults to
-    http://localhost:11434 but can be overridden via the
-    `ollama_host` kwarg (or `ollama_host` in ~/.devgen.yaml).
+    so we reuse the ``openai`` client. The host defaults to
+    http://localhost:11434 but can be overridden via the ``ollama_host``
+    kwarg (or ``ollama_host`` in ~/.devgen.yaml).
     """
 
+    DISPLAY_NAME = "Ollama"
+    REQUIRES_API_KEY = False
     DEFAULT_HOST = "http://localhost:11434"
     DEFAULT_MODEL = "llama3.2"
 
-    def generate(
-        self,
-        prompt: str,
-        api_key: str | None = None,
-        model: str | None = None,
-        **kwargs,
-    ) -> str:
-        from openai import OpenAI
-
+    def _generate(self, prompt, api_key, model, **kwargs):
         host = kwargs.get("ollama_host") or self.DEFAULT_HOST
         base_url = f"{host.rstrip('/')}/v1"
-        chosen_model = model or self.DEFAULT_MODEL
-
-        client = OpenAI(base_url=base_url, api_key=api_key or "ollama")
-
         kwargs.pop("debug", None)
         kwargs.pop("ollama_host", None)
 
-        try:
-            response = client.chat.completions.create(
-                model=chosen_model,
-                messages=[{"role": "user", "content": prompt}],
-                **kwargs,
-            )
-        except Exception as e:
-            if is_token_limit_error(e):
-                raise RuntimeError(format_token_limit_error("Ollama", e)) from e
-            raise RuntimeError(
-                f"Ollama request failed (host={host}, model={chosen_model}): {e}. "
-                "Is the Ollama server running? Start it with `ollama serve` and "
-                "pull a model with `ollama pull <model>`."
-            ) from e
+        client = OpenAI(base_url=base_url, api_key=api_key or "ollama")
+        response = client.chat.completions.create(
+            model=model,
+            messages=[{"role": "user", "content": prompt}],
+            **kwargs,
+        )
 
         if not response.choices:
             raise RuntimeError(
-                f"Ollama returned no choices for model {chosen_model!r}. "
+                f"Ollama returned no choices for model {model!r}. "
                 "The model may not be installed locally."
             )
-        return response.choices[0].message.content.strip()
+        return response.choices[0].message.content
+
+    def _handle_error(self, error):
+        """Override default wrapping — Ollama has no API key to check."""
+        if is_token_limit_error(error):
+            raise RuntimeError(
+                format_token_limit_error(self.DISPLAY_NAME, error)
+            ) from error
+        raise RuntimeError(
+            f"{self.DISPLAY_NAME} request failed: {error}. "
+            "Is the Ollama server running? Start it with `ollama serve` and "
+            "pull a model with `ollama pull <model>`."
+        ) from error

devgen/providers/openai.py

@@ -1,70 +1,20 @@
 from openai import OpenAI
 
-from devgen.utils import format_token_limit_error, is_token_limit_error
+from devgen.providers.base import BaseProvider
 
 
-class OpenaiProvider:
-    """Generates a response string using the OpenAI ChatCompletion API based on the provided prompt and parameters. This method initializes an OpenAI client with the given API key, sends a chat completion request with specified model and additional parameters, and returns the content of the generated message.
-
-    Args:
-        prompt (str): The input prompt to generate a response for.
-        api_key (str): The API key used to authenticate with the OpenAI service.
-        model (str, optional): The model to use for generation; defaults to "gpt-4o".
-        **kwargs: Additional keyword arguments to customize the API request (e.g., temperature).
-
-    Returns:
-        str: The content of the generated response message.
-    """
+class OpenaiProvider(BaseProvider):
+    """Generates content using OpenAI's Chat Completions API."""
 
+    DISPLAY_NAME = "OpenAI"
     DEFAULT_MODEL = "gpt-4o"
 
-    def generate(
-        self, prompt: str, api_key: str, model: str | None = None, **kwargs
-    ) -> str:
-        """Generates a response from the OpenAI ChatCompletion API based on the provided prompt and parameters.
-
-        Creates a client instance with the specified API key, sends a chat completion request using the selected model and additional parameters, and returns the generated message content as a string.
-
-        Args:
-            prompt (str): Prompt input.
-            api_key (str): OpenAI API key.
-            model (str, optional): Model to use (default: gpt-4o).
-            **kwargs: Additional OpenAI ChatCompletion parameters (e.g., temperature).
-
-        Returns:
-            str: Generated response content.
-        """
-        if not api_key:
-            raise ValueError(
-                "OpenAI API key is missing. "
-                "Set it via `devgen setup config` or pass --api-key."
-            )
-
-        # 1. Create a client instance with the API key.
-        try:
-            client = OpenAI(api_key=api_key)
-        except Exception as e:
-            # Add error handling if the client fails to initialize
-            raise RuntimeError(
-                f"Failed to initialize OpenAI client: {e}. "
-                "Verify the key is valid and that the `openai` package is installed."
-            ) from e
-
-        # Remove debug from kwargs if present
+    def _generate(self, prompt, api_key, model, **kwargs):
         kwargs.pop("debug", None)
-
-        # 2. Use the modern API syntax: client.chat.completions.create
-        try:
-            response = client.chat.completions.create(
-                model=model or self.DEFAULT_MODEL,
-                messages=[{"role": "user", "content": prompt}],
-                **kwargs,
-            )
-        except Exception as e:
-            if is_token_limit_error(e):
-                raise RuntimeError(format_token_limit_error("OpenAI", e)) from e
-            raise RuntimeError(
-                f"OpenAI request failed: {e}. "
-                "Check the model name, API key permissions, and your account quota."
-            ) from e
-        return response.choices[0].message.content.strip()
+        client = OpenAI(api_key=api_key)
+        response = client.chat.completions.create(
+            model=model,
+            messages=[{"role": "user", "content": prompt}],
+            **kwargs,
+        )
+        return response.choices[0].message.content

devgen/providers/openrouter.py

@@ -1,46 +1,25 @@
 from openai import OpenAI
 
-from devgen.utils import format_token_limit_error, is_token_limit_error
+from devgen.providers.base import BaseProvider
 
 
-class OpenrouterProvider:
+class OpenrouterProvider(BaseProvider):
     """Generates content using OpenRouter (OpenAI-compatible API)."""
 
+    DISPLAY_NAME = "OpenRouter"
     BASE_URL = "https://openrouter.ai/api/v1"
+    DEFAULT_MODEL = "openai/gpt-3.5-turbo"
 
-    def generate(
-        self, prompt: str, api_key: str, model: str = "openai/gpt-3.5-turbo", **kwargs
-    ) -> str:
-        """Generates a response using OpenRouter."""
-        if not api_key:
-            raise ValueError(
-                "OpenRouter API key is missing. "
-                "Set it via `devgen setup config` or pass --api-key."
-            )
-
-        try:
-            client = OpenAI(
-                base_url=self.BASE_URL,
-                api_key=api_key,
-            )
-
-            # Remove debug from kwargs if present
-            kwargs.pop("debug", None)
-
-            response = client.chat.completions.create(
-                model=model,
-                messages=[{"role": "user", "content": prompt}],
-                extra_headers={
-                    "HTTP-Referer": "https://github.com/S4NKALP/devgen",  # Optional
-                    "X-Title": "devgen CLI",  # Optional
-                },
-                **kwargs,
-            )
-            return response.choices[0].message.content.strip()
-        except Exception as e:
-            if is_token_limit_error(e):
-                raise RuntimeError(format_token_limit_error("OpenRouter", e)) from e
-            raise RuntimeError(
-                f"OpenRouter request failed: {e}. "
-                "Verify the model id (e.g. `openai/gpt-4o`) and your API key credits."
-            ) from e
+    def _generate(self, prompt, api_key, model, **kwargs):
+        kwargs.pop("debug", None)
+        client = OpenAI(base_url=self.BASE_URL, api_key=api_key)
+        response = client.chat.completions.create(
+            model=model,
+            messages=[{"role": "user", "content": prompt}],
+            extra_headers={
+                "HTTP-Referer": "https://github.com/S4NKALP/devgen",
+                "X-Title": "devgen CLI",
+            },
+            **kwargs,
+        )
+        return response.choices[0].message.content