move utils in server

Author: cyclotruc

src/gitingest/config.py

@@ -7,4 +7,5 @@
 MAX_FILES = 10_000  # Maximum number of files to process
 MAX_TOTAL_SIZE_BYTES = 500 * 1024 * 1024  # 500 MB
 
+OUTPUT_FILE_PATH = "digest.txt"
 TMP_BASE_PATH = Path("/tmp/gitingest")

src/server/main.py

@@ -10,9 +10,10 @@
 from slowapi.errors import RateLimitExceeded
 from starlette.middleware.trustedhost import TrustedHostMiddleware
 
+from gitingest.config import TMP_BASE_PATH
 from server.routers import download, dynamic, index
-from server.server_config import DELETE_REPO_AFTER, TMP_BASE_PATH
-from server.server_utils import limiter
+from server.server_config import DELETE_REPO_AFTER, templates
+from server.server_utils import lifespan, limiter, rate_limit_exception_handler
 
 # Load environment variables from .env file
 load_dotenv()
@@ -43,9 +44,6 @@
 # Add middleware to enforce allowed hosts
 app.add_middleware(TrustedHostMiddleware, allowed_hosts=allowed_hosts)
 
-# Set up template rendering
-templates = Jinja2Templates(directory="server/templates")
-
 
 @app.get("/health")
 async def health_check() -> dict[str, str]:

src/server/routers/download.py

@@ -3,7 +3,7 @@
 from fastapi import APIRouter, HTTPException
 from fastapi.responses import Response
 
-from server.server_config import TMP_BASE_PATH
+from gitingest.config import TMP_BASE_PATH
 
 router = APIRouter()
 

src/server/routers/dynamic.py

@@ -4,10 +4,10 @@
 from fastapi.responses import HTMLResponse
 
 from server.query_processor import process_query
+from server.server_config import templates
 from server.server_utils import limiter
 
 router = APIRouter()
-templates = Jinja2Templates(directory="server/templates")
 
 
 @router.get("/{full_path:path}")

src/server/server_config.py

@@ -7,8 +7,6 @@
 MAX_DISPLAY_SIZE: int = 300_000
 DELETE_REPO_AFTER: int = 60 * 60  # In seconds
 
-OUTPUT_FILE_PATH = "digest.txt"
-TMP_BASE_PATH = Path("/tmp/gitingest")
 
 EXAMPLE_REPOS: list[dict[str, str]] = [
     {"name": "Gitingest", "url": "https://github.com/cyclotruc/gitingest"},
@@ -18,4 +16,4 @@
     {"name": "ApiAnalytics", "url": "https://github.com/tom-draper/api-analytics"},
 ]
 
-templates = Jinja2Templates(directory="templates")
+templates = Jinja2Templates(directory="server/templates")

src/server/server_utils.py

@@ -1,14 +1,153 @@
 """ Utility functions for the server. """
 
+import asyncio
 import math
+import shutil
+import time
+from contextlib import asynccontextmanager
+from pathlib import Path
 
-from slowapi import Limiter
+from fastapi import FastAPI, Request
+from fastapi.responses import Response
+from slowapi import Limiter, _rate_limit_exceeded_handler
+from slowapi.errors import RateLimitExceeded
 from slowapi.util import get_remote_address
 
+from gitingest.config import TMP_BASE_PATH
+from server.server_config import DELETE_REPO_AFTER
+
 # Initialize a rate limiter
 limiter = Limiter(key_func=get_remote_address)
 
 
+
+
+
+
+async def rate_limit_exception_handler(request: Request, exc: Exception) -> Response:
+    """
+    Custom exception handler for rate-limiting errors.
+
+    Parameters
+    ----------
+    request : Request
+        The incoming HTTP request.
+    exc : Exception
+        The exception raised, expected to be RateLimitExceeded.
+
+    Returns
+    -------
+    Response
+        A response indicating that the rate limit has been exceeded.
+
+    Raises
+    ------
+    exc
+        If the exception is not a RateLimitExceeded error, it is re-raised.
+    """
+    if isinstance(exc, RateLimitExceeded):
+        # Delegate to the default rate limit handler
+        return _rate_limit_exceeded_handler(request, exc)
+    # Re-raise other exceptions
+    raise exc
+
+
+@asynccontextmanager
+async def lifespan(_: FastAPI):
+    """
+    Lifecycle manager for handling startup and shutdown events for the FastAPI application.
+
+    Parameters
+    ----------
+    _ : FastAPI
+        The FastAPI application instance (unused).
+
+    Yields
+    -------
+    None
+        Yields control back to the FastAPI application while the background task runs.
+    """
+    task = asyncio.create_task(_remove_old_repositories())
+
+    yield
+    # Cancel the background task on shutdown
+    task.cancel()
+    try:
+        await task
+    except asyncio.CancelledError:
+        pass
+
+
+async def _remove_old_repositories():
+    """
+    Periodically remove old repository folders.
+
+    Background task that runs periodically to clean up old repository directories.
+
+    This task:
+    - Scans the TMP_BASE_PATH directory every 60 seconds
+    - Removes directories older than DELETE_REPO_AFTER seconds
+    - Before deletion, logs repository URLs to history.txt if a matching .txt file exists
+    - Handles errors gracefully if deletion fails
+
+    The repository URL is extracted from the first .txt file in each directory,
+    assuming the filename format: "owner-repository.txt"
+    """
+    while True:
+        try:
+            if not TMP_BASE_PATH.exists():
+                await asyncio.sleep(60)
+                continue
+
+            current_time = time.time()
+
+            for folder in TMP_BASE_PATH.iterdir():
+                if folder.is_dir():
+                    continue
+
+                # Skip if folder is not old enough
+                if current_time - folder.stat().st_ctime <= DELETE_REPO_AFTER:
+                    continue
+
+                await _process_folder(folder)
+
+        except Exception as e:
+            print(f"Error in _remove_old_repositories: {e}")
+
+        await asyncio.sleep(60)
+
+
+async def _process_folder(folder: Path) -> None:
+    """
+    Process a single folder for deletion and logging.
+
+    Parameters
+    ----------
+    folder : Path
+        The path to the folder to be processed.
+    """
+    # Try to log repository URL before deletion
+    try:
+        txt_files = [f for f in folder.iterdir() if f.suffix == ".txt"]
+
+        # Extract owner and repository name from the filename
+        if txt_files and "-" in (filename := txt_files[0].stem):
+            owner, repo = filename.split("-", 1)
+            repo_url = f"{owner}/{repo}"
+
+            with open("history.txt", mode="a", encoding="utf-8") as history:
+                history.write(f"{repo_url}\n")
+
+    except Exception as e:
+        print(f"Error logging repository URL for {folder}: {e}")
+
+    # Delete the folder
+    try:
+        shutil.rmtree(folder)
+    except Exception as e:
+        print(f"Error deleting {folder}: {e}")
+
+
 def log_slider_to_size(position: int) -> int:
     """
     Convert a slider position to a file size in bytes using a logarithmic scale.

src/utils.py

@@ -4,8 +4,8 @@
 
 from click.testing import CliRunner
 
-from config import MAX_FILE_SIZE, OUTPUT_FILE_PATH
 from gitingest.cli import main
+from gitingest.config import MAX_FILE_SIZE, OUTPUT_FILE_PATH
 
 
 def test_cli_with_default_options():