feat(logging): implement loguru

Author: MickaelCa

.pre-commit-config.yaml

@@ -117,6 +117,7 @@ repos:
             click>=8.0.0,
             'fastapi[standard]>=0.109.1',
             httpx,
+            loguru>=0.7.0,
             pathspec>=0.12.1,
             prometheus-client,
             pydantic,
@@ -143,6 +144,7 @@ repos:
             click>=8.0.0,
             'fastapi[standard]>=0.109.1',
             httpx,
+            loguru>=0.7.0,
             pathspec>=0.12.1,
             prometheus-client,
             pydantic,

pyproject.toml

@@ -7,6 +7,7 @@ requires-python = ">= 3.8"
 dependencies = [
     "click>=8.0.0",
     "httpx",
+    "loguru>=0.7.0",
     "pathspec>=0.12.1",
     "pydantic",
     "python-dotenv",
@@ -96,7 +97,6 @@ ignore = [  # https://docs.astral.sh/ruff/rules/...
 
     # TODO: fix the following issues:
     "TD003", # missing-todo-link, TODO: add issue links
-    "T201", # print, TODO: replace with logging
     "S108", # hardcoded-temp-file, TODO: replace with tempfile
     "BLE001", # blind-except, TODO: replace with specific exceptions
     "FAST003", # fast-api-unused-path-parameter, TODO: fix

requirements.txt

@@ -2,6 +2,7 @@ boto3>=1.28.0  # AWS SDK for S3 support
 click>=8.0.0
 fastapi[standard]>=0.109.1  # Vulnerable to https://osv.dev/vulnerability/PYSEC-2024-38
 httpx
+loguru>=0.7.0
 pathspec>=0.12.1
 prometheus-client
 pydantic

src/gitingest/ingestion.py

@@ -9,10 +9,14 @@
 from gitingest.output_formatter import format_node
 from gitingest.schemas import FileSystemNode, FileSystemNodeType, FileSystemStats
 from gitingest.utils.ingestion_utils import _should_exclude, _should_include
+from gitingest.utils.logging_config import get_logger
 
 if TYPE_CHECKING:
     from gitingest.schemas import IngestionQuery
 
+# Initialize logger for this module
+logger = get_logger(__name__)
+
 
 def ingest_query(query: IngestionQuery) -> tuple[str, str, str]:
     """Run the ingestion process for a parsed query.
@@ -111,7 +115,14 @@ def _process_node(node: FileSystemNode, query: IngestionQuery, stats: FileSystem
             _process_symlink(path=sub_path, parent_node=node, stats=stats, local_path=query.local_path)
         elif sub_path.is_file():
             if sub_path.stat().st_size > query.max_file_size:
-                print(f"Skipping file {sub_path}: would exceed max file size limit")
+                logger.warning(
+                    "Skipping file: would exceed max file size limit",
+                    extra={
+                        "file_path": str(sub_path),
+                        "file_size": sub_path.stat().st_size,
+                        "max_file_size": query.max_file_size,
+                    },
+                )
                 continue
             _process_file(path=sub_path, parent_node=node, stats=stats, local_path=query.local_path)
         elif sub_path.is_dir():
@@ -133,7 +144,7 @@ def _process_node(node: FileSystemNode, query: IngestionQuery, stats: FileSystem
             node.file_count += child_directory_node.file_count
             node.dir_count += 1 + child_directory_node.dir_count
         else:
-            print(f"Warning: {sub_path} is an unknown file type, skipping")
+            logger.warning("Unknown file type, skipping", extra={"file_path": str(sub_path)})
 
     node.sort_children()
 
@@ -186,12 +197,27 @@ def _process_file(path: Path, parent_node: FileSystemNode, stats: FileSystemStat
 
     """
     if stats.total_files + 1 > MAX_FILES:
-        print(f"Maximum file limit ({MAX_FILES}) reached")
+        logger.warning(
+            "Maximum file limit reached",
+            extra={
+                "current_files": stats.total_files,
+                "max_files": MAX_FILES,
+                "file_path": str(path),
+            },
+        )
         return
 
     file_size = path.stat().st_size
     if stats.total_size + file_size > MAX_TOTAL_SIZE_BYTES:
-        print(f"Skipping file {path}: would exceed total size limit")
+        logger.warning(
+            "Skipping file: would exceed total size limit",
+            extra={
+                "file_path": str(path),
+                "file_size": file_size,
+                "current_total_size": stats.total_size,
+                "max_total_size": MAX_TOTAL_SIZE_BYTES,
+            },
+        )
         return
 
     stats.total_files += 1
@@ -232,15 +258,33 @@ def limit_exceeded(stats: FileSystemStats, depth: int) -> bool:
 
     """
     if depth > MAX_DIRECTORY_DEPTH:
-        print(f"Maximum depth limit ({MAX_DIRECTORY_DEPTH}) reached")
+        logger.warning(
+            "Maximum directory depth limit reached",
+            extra={
+                "current_depth": depth,
+                "max_depth": MAX_DIRECTORY_DEPTH,
+            },
+        )
         return True
 
     if stats.total_files >= MAX_FILES:
-        print(f"Maximum file limit ({MAX_FILES}) reached")
+        logger.warning(
+            "Maximum file limit reached",
+            extra={
+                "current_files": stats.total_files,
+                "max_files": MAX_FILES,
+            },
+        )
         return True  # TODO: end recursion
 
     if stats.total_size >= MAX_TOTAL_SIZE_BYTES:
-        print(f"Maxumum total size limit ({MAX_TOTAL_SIZE_BYTES / 1024 / 1024:.1f}MB) reached")
+        logger.warning(
+            "Maximum total size limit reached",
+            extra={
+                "current_size_mb": stats.total_size / 1024 / 1024,
+                "max_size_mb": MAX_TOTAL_SIZE_BYTES / 1024 / 1024,
+            },
+        )
         return True  # TODO: end recursion
 
     return False

src/gitingest/output_formatter.py

@@ -8,10 +8,14 @@
 
 from gitingest.schemas import FileSystemNode, FileSystemNodeType
 from gitingest.utils.compat_func import readlink
+from gitingest.utils.logging_config import get_logger
 
 if TYPE_CHECKING:
     from gitingest.schemas import IngestionQuery
 
+# Initialize logger for this module
+logger = get_logger(__name__)
+
 _TOKEN_THRESHOLDS: list[tuple[int, str]] = [
     (1_000_000, "M"),
     (1_000, "k"),
@@ -189,8 +193,8 @@ def _format_token_count(text: str) -> str | None:
     try:
         encoding = tiktoken.get_encoding("o200k_base")  # gpt-4o, gpt-4o-mini
         total_tokens = len(encoding.encode(text, disallowed_special=()))
-    except (ValueError, UnicodeEncodeError) as exc:
-        print(exc)
+    except (ValueError, UnicodeEncodeError):
+        logger.exception("Failed to encode text for token counting")
         return None
 
     for threshold, suffix in _TOKEN_THRESHOLDS:

src/gitingest/utils/git_utils.py

@@ -15,11 +15,14 @@
 
 from gitingest.utils.compat_func import removesuffix
 from gitingest.utils.exceptions import InvalidGitHubTokenError
-from server.server_utils import Colors
+from gitingest.utils.logging_config import get_logger
 
 if TYPE_CHECKING:
     from gitingest.schemas import CloneConfig
 
+# Initialize logger for this module
+logger = get_logger(__name__)
+
 # GitHub Personal-Access tokens (classic + fine-grained).
 #   - ghp_ / gho_ / ghu_ / ghs_ / ghr_  → 36 alphanumerics
 #   - github_pat_                       → 22 alphanumerics + "_" + 59 alphanumerics
@@ -97,13 +100,12 @@ async def ensure_git_installed() -> None:
         try:
             stdout, _ = await run_command("git", "config", "core.longpaths")
             if stdout.decode().strip().lower() != "true":
-                print(
-                    f"{Colors.BROWN}WARN{Colors.END}: {Colors.RED}Git clone may fail on Windows "
-                    f"due to long file paths:{Colors.END}",
+                logger.warning(
+                    "Git clone may fail on Windows due to long file paths. "
+                    "Consider enabling long path support with: 'git config --global core.longpaths true'. "
+                    "Note: This command may require administrator privileges.",
+                    extra={"platform": "windows", "longpaths_enabled": False},
                 )
-                print(f"{Colors.RED}To avoid this issue, consider enabling long path support with:{Colors.END}")
-                print(f"{Colors.RED}    git config --global core.longpaths true{Colors.END}")
-                print(f"{Colors.RED}Note: This command may require administrator privileges.{Colors.END}")
         except RuntimeError:
             # Ignore if checking 'core.longpaths' fails.
             pass

src/gitingest/utils/logging_config.py

@@ -0,0 +1,115 @@
+"""Logging configuration for gitingest using loguru.
+
+This module provides structured JSON logging suitable for Kubernetes deployments
+while also supporting human-readable logging for development.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from typing import Any
+
+from loguru import logger
+
+
+def serialize_record(record: dict[str, Any]) -> str:
+    """Serialize log record for structured logging.
+
+    Parameters
+    ----------
+    record : dict[str, Any]
+        The log record dictionary
+
+    Returns
+    -------
+    str
+        Serialized log record
+
+    """
+    # Extract relevant fields for structured logging
+    log_entry = {
+        "timestamp": record["time"].isoformat(),
+        "level": record["level"].name,
+        "logger": record["name"],
+        "module": record["module"],
+        "function": record["function"],
+        "line": record["line"],
+        "message": record["message"],
+    }
+
+    # Add exception info if present
+    if record["exception"]:
+        log_entry["exception"] = {
+            "type": record["exception"].type.__name__,
+            "value": str(record["exception"].value),
+            "traceback": record["exception"].traceback,
+        }
+
+    # Add extra fields if present
+    if hasattr(record, "extra"):
+        log_entry["extra"] = record["extra"]
+
+    return str(log_entry)
+
+
+def configure_logging() -> None:
+    """Configure loguru for the application.
+
+    Sets up JSON logging for production/Kubernetes environments
+    or human-readable logging for development.
+    """
+    # Remove default handler
+    logger.remove()
+
+    # Check if we're in Kubernetes or production environment
+    is_k8s = os.getenv("KUBERNETES_SERVICE_HOST") is not None
+    log_format = os.getenv("LOG_FORMAT", "json" if is_k8s else "human")
+    log_level = os.getenv("LOG_LEVEL", "INFO")
+
+    if log_format.lower() == "json":
+        # JSON format for structured logging (Kubernetes/production)
+        logger.add(
+            sys.stdout,
+            format=serialize_record,
+            level=log_level,
+            enqueue=True,  # Async logging for better performance
+            diagnose=False,  # Don't include variable values in exceptions (security)
+            backtrace=True,  # Include full traceback
+        )
+    else:
+        # Human-readable format for development
+        logger.add(
+            sys.stdout,
+            format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | "
+            "<level>{level: <8}</level> | "
+            "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> | "
+            "<level>{message}</level>",
+            level=log_level,
+            enqueue=True,
+            diagnose=True,  # Include variable values in development
+            backtrace=True,
+        )
+
+
+def get_logger(name: str | None = None) -> logger.__class__:
+    """Get a configured logger instance.
+
+    Parameters
+    ----------
+    name : str | None, optional
+        Logger name, defaults to the calling module name
+
+    Returns
+    -------
+    logger.__class__
+        Configured logger instance
+
+    """
+    if name:
+        return logger.bind(name=name)
+    return logger
+
+
+# Initialize logging when module is imported
+configure_logging()

src/server/metrics_server.py

@@ -1,14 +1,14 @@
 """Prometheus metrics server running on a separate port."""
 
-import logging
-
 import uvicorn
 from fastapi import FastAPI
 from fastapi.responses import HTMLResponse
 from prometheus_client import REGISTRY, generate_latest
 
+from gitingest.utils.logging_config import get_logger
+
 # Create a logger for this module
-logger = logging.getLogger(__name__)
+logger = get_logger(__name__)
 
 # Create a separate FastAPI app for metrics
 metrics_app = FastAPI(