Enhance code quality of async methods used for repo processing.

readmeai/config/settings.py

@@ -4,7 +4,7 @@
 from enum import Enum
 from importlib import resources
 from pathlib import Path
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Union
 from urllib.parse import urlparse, urlsplit
 
 import pkg_resources
@@ -99,59 +99,57 @@ class FileSettings(BaseModel):
 
 
 class GitSettings(BaseModel):
-    """Pydantic model for Git repository details."""
+    """Codebase repository settings and validations."""
 
-    repository: str
+    repository: Union[str, Path]
     source: Optional[str]
     name: Optional[str]
 
     @validator("repository", pre=True, always=True)
-    def validate_repository(cls, value: str) -> str:
+    def validate_repository(cls, value: Union[str, Path]) -> Union[str, Path]:
         """Validate the repository URL or path."""
-        path = Path(value)
-        if path.is_dir():
+        if isinstance(value, str):
+            path = Path(value)
+            if path.is_dir():
+                return value
+            try:
+                parsed_url = urlparse(value)
+                if parsed_url.scheme in ["http", "https"] and any(
+                    service.host in parsed_url.netloc for service in GitService
+                ):
+                    return value
+            except ValueError:
+                pass
+        elif isinstance(value, Path) and value.is_dir():
             return value
-        try:
-            parsed_url = urlparse(value)
-        except ValueError:
-            raise ValueError(f"Invalid repository URL or path: {value}")
-
-        if parsed_url.scheme != "https" or not any(
-            service.host in parsed_url.netloc for service in GitService
-        ):
-            raise ValueError(f"Invalid repository URL or path: {value}")
-
-        return value
+        raise ValueError(f"Invalid repository URL or path: {value}")
 
     @validator("source", pre=True, always=True)
-    def set_source(cls, value: str, values: dict) -> str:
+    def set_source(cls, value: Optional[str], values: dict) -> str:
         """Sets the Git service source from the repository provided."""
         repo = values.get("repository")
-
-        if Path(repo).is_dir():
+        if isinstance(repo, Path) or (
+            isinstance(repo, str) and Path(repo).is_dir()
+        ):
             return GitService.LOCAL.host
 
-        parsed_url = urlparse(repo)
+        parsed_url = urlparse(str(repo))
         for service in GitService:
             if service.host in parsed_url.netloc:
                 return service.host
-
-        raise ValueError("Unsupported Git service.")
+        return GitService.LOCAL.host
 
     @validator("name", pre=True, always=True)
-    def set_name(cls, value: str, values: dict) -> str:
+    def set_name(cls, value: Optional[str], values: dict) -> str:
         """Sets the repository name from the repository provided."""
         repo = values.get("repository")
-        parsed_url = urlsplit(repo)
-        for service in GitService:
-            if service.host in parsed_url.netloc:
-                path = parsed_url.path
-                name = path.rsplit("/", 1)[-1] if "/" in path else path
-                if name.endswith(".git"):
-                    name = name[:-4]
-                return name
-
-        return Path(repo).name
+        if isinstance(repo, Path):
+            return repo.name
+        elif isinstance(repo, str):
+            parsed_url = urlsplit(repo)
+            name = parsed_url.path.split("/")[-1]
+            return name.removesuffix(".git")
+        return "n/a"
 
 
 class LlmApiSettings(BaseModel):

readmeai/core/model.py

@@ -39,7 +39,7 @@ class LlmApiHandler:
     async def use_api(self) -> None:
         """Context manager for manage resources used by the HTTP client."""
         try:
-            yield
+            yield self
         finally:
             await self.close()
 

readmeai/main.py

@@ -6,7 +6,10 @@
 
 import asyncio
 import os
+import shutil
+import tempfile
 import traceback
+from pathlib import Path
 
 from readmeai.cli.options import prompt_for_custom_image
 from readmeai.config.settings import (
@@ -30,16 +33,18 @@ async def readme_agent(conf: AppConfig, conf_helper: ConfigHelper) -> None:
     logger.info(f"Processing repository: {conf.git.repository}")
     logger.info(f"GPT language model engine: {conf.llm.model}")
 
+    llm = model.LlmApiHandler(conf)
+    repo_name = conf.git.name.upper()
+    repo_url = conf.git.repository
+    temp_dir = await asyncio.to_thread(tempfile.mkdtemp)
+
     try:
-        llm = model.LlmApiHandler(conf)
-        repo_name = conf.git.name.upper()
-        repo_url = conf.git.repository
-        temp_dir = clone_repo_to_temp_dir(repo_url)
+        await clone_repo_to_temp_dir(repo_url, temp_dir)
 
         conf.md.tree = conf.md.tree.format(
             tree.TreeGenerator(
                 conf_helper=conf_helper,
-                root_directory=temp_dir,
+                root_directory=Path(temp_dir),
                 repo_url=repo_url,
                 repo_name=repo_name,
             ).generate_and_format_tree()
@@ -51,27 +56,27 @@ async def readme_agent(conf: AppConfig, conf_helper: ConfigHelper) -> None:
             for file_path, file_content in file_context.items()
         ]
 
-        logger.info(f"Repository dependencies and software: {dependencies}")
-        logger.info(f"Repository directory structure:\n{conf.md.tree}")
+        logger.info(f"Dependencies and software: {dependencies}")
+        logger.info(f"Directory tree structure:\n{conf.md.tree}")
 
-        async with llm.use_api():
+        async with llm.use_api() as api:
             context = {
                 "repo": repo_url,
                 "tree": conf.md.tree,
                 "dependencies": dependencies,
                 "summaries": summaries,
             }
             if conf.cli.offline is False:
-                context["summaries"] = await llm.prompt_processor(
+                context["summaries"] = await api.prompt_processor(
                     "summaries", context
                 )
-                features_response = await llm.prompt_processor(
+                features_response = await api.prompt_processor(
                     "features", context
                 )
-                overview_response = await llm.prompt_processor(
+                overview_response = await api.prompt_processor(
                     "overview", context
                 )
-                slogan_response = await llm.prompt_processor("slogan", context)
+                slogan_response = await api.prompt_processor("slogan", context)
                 summaries = context["summaries"]
                 conf.md.features = conf.md.features.format(
                     features_response["features"]
@@ -81,9 +86,7 @@ async def readme_agent(conf: AppConfig, conf_helper: ConfigHelper) -> None:
                 )
                 conf.md.slogan = slogan_response["slogan"]
             else:
-                logger.warning(
-                    "Offline mode enabled. Skipping LLM API prompts."
-                )
+                logger.warning("Offline mode enabled, skipping API calls.")
                 (
                     summaries,
                     conf.md.features,
@@ -99,18 +102,11 @@ async def readme_agent(conf: AppConfig, conf_helper: ConfigHelper) -> None:
                     conf.md.default,
                 )
 
-        logger.info(f"Total summaries generated: {len(summaries)}")
-
         headers.build_readme_md(conf, conf_helper, dependencies, summaries)
+        logger.info("README file successfully created: {conf.files.output}")
 
-        logger.info(
-            "README-AI file generated successfully: {conf.files.output}"
-        )
-
-    except Exception as exc_info:
-        logger.error(
-            f"Exception occurred: {exc_info}\n{traceback.format_exc()}"
-        )
+    finally:
+        shutil.rmtree(temp_dir)
 
 
 def main(
@@ -127,28 +123,39 @@ def main(
     temperature: float,
 ) -> None:
     """Main method of the readme-ai CLI application."""
-    logger.info("Executing the README-AI CLI application.")
-    conf = load_config()
-    conf_model = AppConfigModel(app=conf)
-    conf_helper = load_config_helper(conf_model)
-    conf.git = GitSettings(repository=repository)
-    conf.files.output = output
-    conf.cli.emojis = emojis
-    conf.cli.offline = offline
-    conf.llm.tokens_max = max_tokens
-    conf.llm.model = model
-    conf.llm.temperature = temperature
-    conf.md.align = align
-    conf.md.badges_style = badges
-
-    if image == ImageOptions.CUSTOM.name:
-        conf.md.image = prompt_for_custom_image(None, None, image)
-    else:
-        conf.md.image = image
+    try:
+        conf = load_config()
+        conf_model = AppConfigModel(app=conf)
+        conf_helper = load_config_helper(conf_model)
+        conf.git = GitSettings(repository=repository)
+        conf.files.output = output
+        conf.cli.emojis = emojis
+        conf.cli.offline = offline
+        conf.llm.tokens_max = max_tokens
+        conf.llm.model = model
+        conf.llm.temperature = temperature
+        conf.md.align = align
+        conf.md.badges_style = badges
+
+        if image == ImageOptions.CUSTOM.name:
+            conf.md.image = prompt_for_custom_image(None, None, image)
+        else:
+            conf.md.image = image
+
+        _set_environment_vars(conf, api_key)
+
+        asyncio.run(readme_agent(conf, conf_helper))
+
+    except Exception as exc_info:
+        logger.error(
+            f"Exception during README generation: {exc_info}\n{traceback.format_exc()}"
+        )
+
 
+def _set_environment_vars(config: AppConfig, api_key: str) -> None:
+    """Set environment variables for the CLI application."""
     if api_key is not None:
         os.environ["OPENAI_API_KEY"] = api_key
-    elif "OPENAI_API_KEY" not in os.environ:
-        conf.cli.offline = True
 
-    asyncio.run(readme_agent(conf, conf_helper))
+    elif "OPENAI_API_KEY" not in os.environ:
+        config.cli.offline = True

readmeai/services/git_operations.py

@@ -3,22 +3,23 @@
 import os
 import platform
 import shutil
-import tempfile
 from pathlib import Path
 from typing import Optional
 
 import git
 
 
-def clone_repo_to_temp_dir(repo_path: str) -> Path:
+async def clone_repo_to_temp_dir(repo_url: str, temp_dir: str) -> str:
     """Clone the repository to a temporary directory."""
-    if Path(repo_path).exists():
-        return Path(repo_path)
-
     try:
-        temp_dir = tempfile.mkdtemp()
-        git.Repo.clone_from(repo_path, temp_dir, depth=1, single_branch=True)
-        return Path(temp_dir)
+        repo_path = Path(repo_url)
+        if repo_path.is_dir():
+            shutil.copytree(repo_url, temp_dir, dirs_exist_ok=True)
+        else:
+            git.Repo.clone_from(
+                repo_url, temp_dir, depth=1, single_branch=True
+            )
+        return temp_dir
 
     except git.GitCommandError as exc_info:
         raise ValueError(f"Git clone error: {exc_info}") from exc_info

tests/test_services/test_git_operations.py

@@ -1,6 +1,6 @@
 """Unit tests for Git operation utility methods."""
 
-import shutil
+import os
 import tempfile
 from pathlib import Path
 from unittest import mock
@@ -14,21 +14,44 @@
     validate_git_executable,
 )
 
+TEST_REPO_URL = "https://github.com/eli64s/readme-ai-streamlit"
 
-@pytest.fixture
-def temp_dir():
-    """Returns a temporary directory."""
-    dir = tempfile.mkdtemp()
-    yield Path(dir)
-    shutil.rmtree(dir)
 
+@pytest.mark.asyncio
+async def test_clone_valid_repo():
+    """Test that a valid repository is cloned."""
+    with tempfile.TemporaryDirectory() as temp_dir:
+        cloned_dir = await clone_repo_to_temp_dir(TEST_REPO_URL, temp_dir)
+        assert os.path.isdir(cloned_dir)
 
-def test_clone_repo_to_temp_dir(temp_dir):
-    """Test that the repo is cloned to a temporary directory."""
-    repo = "https://github.com/eli64s/readme-ai-streamlit"
-    cloned_dir = clone_repo_to_temp_dir(repo)
-    assert isinstance(cloned_dir, Path)
-    assert cloned_dir.exists()
+
+@pytest.mark.asyncio
+async def test_clone_invalid_repo():
+    """Test that an invalid repository is not cloned."""
+    with tempfile.TemporaryDirectory() as temp_dir:
+        with pytest.raises(ValueError):
+            await clone_repo_to_temp_dir("https://invalid/repo.git", temp_dir)
+
+
+@pytest.mark.asyncio
+async def test_clone_local_repo():
+    """Test that a local repository is cloned."""
+    with tempfile.TemporaryDirectory() as source_dir:
+        os.mkdir(
+            os.path.join(source_dir, ".git")
+        )  # Mock a local git repository
+        with tempfile.TemporaryDirectory() as temp_dir:
+            cloned_dir = await clone_repo_to_temp_dir(source_dir, temp_dir)
+            assert os.path.isdir(cloned_dir)
+            assert os.path.exists(os.path.join(cloned_dir, ".git"))
+
+
+@pytest.mark.asyncio
+async def test_clone_nonexistent_local_path():
+    """Test that a nonexistent local path is not cloned."""
+    with tempfile.TemporaryDirectory() as temp_dir:
+        with pytest.raises(ValueError):
+            await clone_repo_to_temp_dir("/nonexistent/path", temp_dir)
 
 
 def test_find_git_executable():