contract_reporter.py

Code Hygiene Score: 78

Keine Issues gefunden.

Dependencies 14

Klassen 1

Code

"""Contract Reporting and Logging Module."""

import json
import time
from datetime import datetime
from typing import Any, Callable, Optional

from config import Config
from shared.domain import LogEntry
from shared.infrastructure import get_logger

from .constants import (
    ERROR_MESSAGE_MAX_LENGTH,
    LOG_STATUS_DENIED,
    LOG_STATUS_ERROR,
    LOG_STATUS_SUCCESS,
    MS_PER_SECOND,
)


class ContractReporter:
    """Handles logging and response formatting for contract operations."""

    def __init__(self):
        """Initialize reporter with logger."""
        self.logger = get_logger("mcp-contracts", Config)

    def log_operation(
        self,
        tool_name: str,
        request_data: dict[str, Any],
        status: str,
        duration_ms: int,
        error_message: Optional[str] = None,
    ) -> None:
        """
        Log a contract operation.

        Args:
            tool_name: Name of the tool/operation
            request_data: Request parameters as dict
            status: Operation status (success, error, denied)
            duration_ms: Duration in milliseconds
            error_message: Optional error message
        """
        request_str = json.dumps(request_data)
        error_truncated = None
        if error_message:
            error_truncated = error_message[:ERROR_MESSAGE_MAX_LENGTH]

        log_entry = LogEntry(
            timestamp=datetime.now(),
            client_name="mcp-contracts",
            tool_name=tool_name,
            request=request_str,
            status=status,
            duration_ms=duration_ms,
            error_message=error_truncated,
        )
        self.logger.log(log_entry)

    def execute_with_logging(
        self,
        tool_name: str,
        request_data: dict[str, Any],
        operation: Callable[[], dict[str, Any]],
    ) -> dict[str, Any]:
        """
        Execute an operation with automatic logging.

        Args:
            tool_name: Name of the tool/operation
            request_data: Request parameters as dict
            operation: Callable that executes the operation

        Returns:
            Operation result dictionary
        """
        start = time.time()

        try:
            result = operation()
            duration = int((time.time() - start) * MS_PER_SECOND)

            # Log success
            self.log_operation(
                tool_name=tool_name,
                request_data=request_data,
                status=LOG_STATUS_SUCCESS,
                duration_ms=duration,
            )

            return result

        except Exception as e:
            duration = int((time.time() - start) * MS_PER_SECOND)

            # Log error
            self.log_operation(
                tool_name=tool_name,
                request_data=request_data,
                status=LOG_STATUS_ERROR,
                duration_ms=duration,
                error_message=str(e),
            )

            return {"success": False, "error": str(e)}

    def log_not_found(
        self,
        tool_name: str,
        request_data: dict[str, Any],
        message: str = "Contract not found",
    ) -> None:
        """
        Log a not-found error.

        Args:
            tool_name: Name of the tool/operation
            request_data: Request parameters as dict
            message: Error message
        """
        self.log_operation(
            tool_name=tool_name,
            request_data=request_data,
            status=LOG_STATUS_DENIED,
            duration_ms=0,
            error_message=message,
        )

    @staticmethod
    def format_success_response(data: dict[str, Any]) -> dict[str, Any]:
        """
        Format a successful response.

        Args:
            data: Response data

        Returns:
            Formatted response with success flag
        """
        return {"success": True, **data}

    @staticmethod
    def format_error_response(error: str) -> dict[str, Any]:
        """
        Format an error response.

        Args:
            error: Error message

        Returns:
            Formatted error response
        """
        return {"success": False, "error": error}

    @staticmethod
    def format_list_response(
        items: list[Any],
        total: int,
        limit: int,
        compact: bool = True,
    ) -> dict[str, Any]:
        """
        Format a list response.

        Args:
            items: List of items (with to_dict or to_dict_compact methods)
            total: Total count
            limit: Limit used
            compact: Whether to use compact format

        Returns:
            Formatted list response
        """
        formatted_items = []
        for item in items:
            if hasattr(item, "to_dict_compact") and compact:
                formatted_items.append(item.to_dict_compact())
            elif hasattr(item, "to_dict"):
                formatted_items.append(item.to_dict())
            else:
                formatted_items.append(item)

        return ContractReporter.format_success_response(
            {
                "contracts": formatted_items,
                "total": total,
                "limit": limit,
            }
        )
← Übersicht Graph