Reduce complexity of various methods.

readmeai/cli/options.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import os
+from typing import Optional
 
 import click
 from click import Context, Parameter
@@ -11,9 +12,9 @@
 
 
 def prompt_for_custom_image(
-    context: Context | None,
-    parameter: Parameter | None,
-    value: str | None,
+    context: Optional[Context],
+    parameter: Optional[Parameter],
+    value: Optional[str],
 ) -> str:
     """Prompt the user for a custom image URL."""
     if value == ImageOptions.CUSTOM.name:
@@ -40,9 +41,7 @@ def prompt_for_custom_image(
 badges = click.option(
     "-b",
     "--badges",
-    type=click.Choice(
-        [opt.value for opt in BadgeOptions], case_sensitive=False
-    ),
+    type=click.Choice([opt.value for opt in BadgeOptions], case_sensitive=False),
     default=BadgeOptions.DEFAULT.value,
     help="""\
         Badge icon style types to select from when generating README.md badges. The following options are currently available:\n
@@ -66,9 +65,7 @@ def prompt_for_custom_image(
 image = click.option(
     "-i",
     "--image",
-    type=click.Choice(
-        [opt.name for opt in ImageOptions], case_sensitive=False
-    ),
+    type=click.Choice([opt.name for opt in ImageOptions], case_sensitive=False),
     default=ImageOptions.DEFAULT.name,
     callback=prompt_for_custom_image,
     show_choices=True,

readmeai/core/model.py

@@ -5,8 +5,14 @@
 from contextlib import asynccontextmanager
 from typing import Any, Dict, List, Tuple, Union
 
-import aiohttp
 import openai
+from httpx import (
+    AsyncClient,
+    HTTPStatusError,
+    Limits,
+    NetworkError,
+    TimeoutException,
+)
 from tenacity import (
     retry,
     retry_if_exception_type,
@@ -32,20 +38,28 @@ def __init__(self, config: settings.AppConfig) -> None:
         """Initializes the GPT language model API handler."""
         self.cache = {}
         self.config = config
-        self.encoder = config.llm.encoding
         self.logger = Logger(__name__)
-        self.prompts = config.prompts
-        self.tokens = config.llm.tokens
-        self.tokens_max = config.llm.tokens_max
-        self.http_client = aiohttp.ClientSession(
-            timeout=aiohttp.ClientTimeout(total=20),
-            connector=aiohttp.TCPConnector(
-                limit=200, limit_per_host=100, enable_cleanup_closed=True
-            ),
-        )
-        self.rate_limit_semaphore = asyncio.Semaphore(config.llm.rate_limit)
+        self._llm_attributes()
+        self._http_client()
         self._handle_response = functools.lru_cache(maxsize=100)(self._handle_response)
 
+    def _llm_attributes(self):
+        """Initializes basic attributes for the class."""
+        self.encoder = self.config.llm.encoding
+        self.prompts = self.config.prompts
+        self.tokens = self.config.llm.tokens
+        self.tokens_max = self.config.llm.tokens_max
+        self.rate_limit = self.config.llm.rate_limit
+
+    def _http_client(self):
+        """Configures the HTTP client for the class."""
+        self.http_client = AsyncClient(
+            http2=True,
+            timeout=20,
+            limits=Limits(max_keepalive_connections=100, max_connections=20),
+        )
+        self.rate_limit_semaphore = asyncio.Semaphore(self.rate_limit)
+
     @asynccontextmanager
     async def use_api(self) -> None:
         """Context manager for HTTP client used by the LLM API."""
@@ -56,7 +70,7 @@ async def use_api(self) -> None:
 
     async def close(self) -> None:
         """Closes the HTTP client."""
-        await self.http_client.close()
+        await self.http_client.aclose()
 
     async def batch_request(
         self,
@@ -66,21 +80,42 @@ async def batch_request(
     ) -> List[str]:
         """Generates text for the README.md file using GPT language models."""
         prompts = await self._set_prompt_context(file_context, dependencies, summaries)
+        responses = await self._batch_prompts(prompts)
+        return responses
+
+    async def _batch_prompts(
+        self, prompts: List[Union[str, Tuple[str, str]]], batch_size: int = 5
+    ):
+        """Processes prompts in batches and returns the generated text."""
         responses = []
-        for batch in self._batch_prompts(prompts):
+
+        for batch in self._generate_batches(prompts, batch_size):
             batch_responses = await asyncio.gather(
-                *[self._process_prompt(prompt) for prompt in batch]
+                *[self._process_batch(prompt) for prompt in batch]
                 # , return_exceptions=True
             )
             responses.extend(batch_responses)
+
         return responses
 
-    def _batch_prompts(
-        self, prompts: List[Union[str, Tuple[str, str]]], batch_size: int = 5
-    ) -> List[List[Union[str, Tuple[str, str]]]]:
-        """Batches prompts for the LLM API."""
-        for i in range(0, len(prompts), batch_size):
-            yield prompts[i : i + batch_size]
+    def _generate_batches(self, items: List[Any], batch_size: int):
+        """Generator to create batches from a list of items."""
+        for i in range(0, len(items), batch_size):
+            yield items[i : i + batch_size]
+
+    async def _process_batch(self, prompt: Dict[str, Any]) -> str:
+        """Processes a prompt and returns the generated text."""
+        if prompt["type"] == "summaries":
+            return await self._handle_code_summary_response(prompt["context"])
+        else:
+            formatted_prompt = self._get_prompt_context(
+                prompt["type"], prompt["context"]
+            )
+            tokens = adjust_max_tokens(self.tokens, formatted_prompt)
+            _, summary = await self._handle_response(
+                prompt["type"], formatted_prompt, tokens
+            )
+            return summary
 
     async def _set_prompt_context(
         self,
@@ -128,35 +163,31 @@ async def _set_prompt_context(
             ]
         ]
 
-    def _inject_prompt_context(self, prompt_type, context) -> str:
+    def _get_prompt_context(self, prompt_type, context) -> str:
         """Generates a prompt for the LLM API."""
+        prompt_template = self._get_prompt_template(prompt_type)
+        if not prompt_template:
+            self.logger.error(f"Prompt type '{prompt_type}' not found.")
+            return ""
+        return self._inject_prompt_context(prompt_template, context)
+
+    def _get_prompt_template(self, prompt_type: str) -> str:
+        """Retrieves the template for the given prompt type."""
         prompt_templates = {
             "features": self.prompts.features,
             "overview": self.prompts.overview,
             "slogan": self.prompts.slogan,
         }
-        prompt_template = prompt_templates.get(prompt_type)
+        return prompt_templates.get(prompt_type, "")
 
-        if prompt_template:
-            return prompt_template.format(*[context[key] for key in context])
-        else:
-            self.logger.error(f"Unknown prompt type: {prompt_type}")
+    def _inject_prompt_context(self, template: str, context: dict) -> str:
+        """Formats the template with the provided context."""
+        try:
+            return template.format(*[context[key] for key in context])
+        except KeyError as exc:
+            self.logger.error(f"Missing context for prompt key: {exc}")
             return ""
 
-    async def _process_prompt(self, prompt: Dict[str, Any]) -> str:
-        """Processes a prompt and returns the generated text."""
-        if prompt["type"] == "summaries":
-            return await self._handle_code_summary_response(prompt["context"])
-        else:
-            formatted_prompt = self._inject_prompt_context(
-                prompt["type"], prompt["context"]
-            )
-            tokens = adjust_max_tokens(self.tokens, formatted_prompt)
-            _, summary = await self._handle_response(
-                prompt["type"], formatted_prompt, tokens
-            )
-            return summary
-
     async def _handle_code_summary_response(
         self, file_context: List[Tuple[str, str]]
     ) -> List[Tuple[str, str]]:
@@ -190,9 +221,9 @@ async def _handle_code_summary_response(
         wait=wait_exponential(multiplier=1, min=2, max=6),
         retry=retry_if_exception_type(
             (
-                aiohttp.ClientConnectionError,
-                aiohttp.ClientResponseError,
-                aiohttp.ServerTimeoutError,
+                HTTPStatusError,
+                NetworkError,
+                TimeoutException,
                 openai.error.OpenAIError,
             )
         ),
@@ -228,24 +259,25 @@ async def _handle_response(
             prompt = truncate_tokens(self.encoder, prompt, tokens)
 
         try:
-            async with self.rate_limit_semaphore, self.http_client.post(
-                self.config.llm.endpoint,
-                headers={"Authorization": f"Bearer {openai.api_key}"},
-                json={
-                    "messages": [
-                        {
-                            "role": "system",
-                            "content": self.config.llm.content,
-                        },
-                        {"role": "user", "content": prompt},
-                    ],
-                    "model": self.config.llm.model,
-                    "temperature": self.config.llm.temperature,
-                    "max_tokens": tokens,
-                },
-            ) as response:
+            async with self.rate_limit_semaphore:
+                response = await self.http_client.post(
+                    self.config.llm.endpoint,
+                    headers={"Authorization": f"Bearer {openai.api_key}"},
+                    json={
+                        "messages": [
+                            {
+                                "role": "system",
+                                "content": self.config.llm.content,
+                            },
+                            {"role": "user", "content": prompt},
+                        ],
+                        "model": self.config.llm.model,
+                        "temperature": self.config.llm.temperature,
+                        "max_tokens": tokens,
+                    },
+                )
                 response.raise_for_status()
-                llm_response = await response.json()
+                llm_response = response.json()
                 llm_text = llm_response["choices"][0]["message"]["content"]
                 llm_text = (
                     format_sentence(llm_text)
@@ -257,9 +289,9 @@ async def _handle_response(
                 return index, llm_text
 
         except (
-            aiohttp.ClientConnectionError,
-            aiohttp.ClientResponseError,
-            aiohttp.ServerTimeoutError,
+            HTTPStatusError,
+            NetworkError,
+            TimeoutException,
             openai.error.OpenAIError,
         ) as exc:
             error_msg = f"Error generating text for {index}: {exc}"

readmeai/exceptions.py

@@ -6,33 +6,23 @@
 class ReadmeAiException(Exception):
     """Base exception for the readme-ai application."""
 
-    pass
+    ...
 
 
-class RepositoryError(ReadmeAiException):
-    """Exceptions related to repository operations."""
-
-    pass
-
-
-class GitCloneError(RepositoryError):
+class GitCloneError(ReadmeAiException):
     """Could not clone repository."""
 
     def __init__(self, repository: str, *args):
         self.repository = repository
         super().__init__(f"Failed to clone repository: {repository}", *args)
 
 
-class ReadmeGenerationError(ReadmeAiException):
+class ReadmeGeneratorError(ReadmeAiException):
     """Exceptions related to readme generation."""
 
-    pass
-
-
-class ApiCommunicationError(ReadmeAiException):
-    """Exceptions related to external APIs."""
-
-    pass
+    def __init__(self, traceback, *args):
+        self.traceback = traceback
+        super().__init__(f"Error generating readme: {traceback}", *args)
 
 
 class FileSystemError(ReadmeAiException):
@@ -46,10 +36,10 @@ def __init__(self, message, path, *args):
 class FileReadError(FileSystemError):
     """Could not read file."""
 
-    pass
+    ...
 
 
 class FileWriteError(FileSystemError):
     """Could not write file."""
 
-    pass
+    ...

readmeai/main.py

@@ -23,7 +23,7 @@
 from readmeai.core.logger import Logger
 from readmeai.core.model import ModelHandler
 from readmeai.core.preprocess import FileData, process_repository
-from readmeai.exceptions import ReadmeGenerationError
+from readmeai.exceptions import ReadmeGeneratorError
 from readmeai.markdown.builder import build_readme_md
 from readmeai.services.git_utils import clone_to_temporary_directory
 
@@ -82,7 +82,7 @@ async def readme_agent(conf: AppConfig, conf_helper: ConfigHelper) -> None:
         logger.info(f"README.md file generated successfully @ {conf.files.output}")
 
     except Exception as exc:
-        raise ReadmeGenerationError(exc, traceback.format_exc()) from exc
+        raise ReadmeGeneratorError(traceback.format_exc()) from exc
 
 
 def main(
@@ -124,7 +124,7 @@ def main(
         asyncio.run(readme_agent(conf, conf_helper))
 
     except Exception as exc:
-        raise ReadmeGenerationError(exc, traceback.format_exc()) from exc
+        raise ReadmeGeneratorError(exc, traceback.format_exc()) from exc
 
 
 def setup_environment(config: AppConfig, api_key: str) -> None:

readmeai/services/git_metadata.py

@@ -77,7 +77,7 @@ async def _fetch_git_metadata(
 
 async def git_api_request(
     session: aiohttp.ClientSession, repo_url: str
-) -> GitHubRepoMetadata | None:
+) -> Optional[GitHubRepoMetadata]:
     """Retrieves repo metadata and returns a GitHubRepoMetadata instance."""
     api_url = await fetch_git_api_url(repo_url)
     if not api_url:

readmeai/services/git_utils.py

@@ -4,6 +4,7 @@
 import platform
 import shutil
 from pathlib import Path
+from typing import Optional
 
 import git
 
@@ -67,7 +68,7 @@ def fetch_git_file_url(file_path: str, full_name: str, repo_url: str) -> str:
     return file_path
 
 
-def find_git_executable() -> Path | None:
+def find_git_executable() -> Optional[Path]:
     """Find the path to the git executable, if available."""
     git_exec_path = os.environ.get("GIT_PYTHON_GIT_EXECUTABLE")
     if git_exec_path:
@@ -102,7 +103,7 @@ def validate_file_permissions(temp_dir: Path) -> None:
             )
 
 
-def validate_git_executable(git_exec_path: str | None) -> None:
+def validate_git_executable(git_exec_path: str) -> None:
     """Validate the path to the git executable."""
     if not git_exec_path or not Path(git_exec_path).exists():
         raise ValueError(f"Git executable not found at {git_exec_path}")