get_history

Instructions

Implementation Reference

• src/csv_editor/server.py:493-498 (handler)

  MCP tool registration for 'get_history' - a FastMCP @mcp.tool decorator that delegates to _get_operation_history.

  @mcp.tool
  async def get_history(
      session_id: str, limit: int | None = None, ctx: Context = None
  ) -> dict[str, Any]:
      """Get operation history for a session."""
      return await _get_operation_history(session_id, limit, ctx)

• src/csv_editor/tools/history_operations.py:110-158 (handler)

  Handler function that resolves the session and calls session.get_history(limit) to retrieve operation history.

  async def get_operation_history(
      session_id: str, limit: int | None = None, ctx: Context = None
  ) -> dict[str, Any]:
      """
      Get operation history for a session.
  
      Args:
          session_id: Session identifier
          limit: Maximum number of operations to return
          ctx: FastMCP context
  
      Returns:
          Dict with history and statistics
      """
      try:
          manager = get_session_manager()
          session = manager.get_session(session_id)
  
          if not session:
              return OperationResult(
                  success=False,
                  message="Session not found",
                  error=f"No session with ID: {session_id}",
              ).model_dump()
  
          if ctx:
              await ctx.info(f"Getting operation history for session {session_id}")
  
          result = session.get_history(limit)
  
          if result["success"]:
              return OperationResult(
                  success=True,
                  message="History retrieved successfully",
                  session_id=session_id,
                  data=result,
              ).model_dump()
          else:
              return OperationResult(
                  success=False, message="Failed to get history", error=result.get("error")
              ).model_dump()
  
      except Exception as e:
          logger.error(f"Error getting history: {e!s}")
          if ctx:
              await ctx.error(f"Failed to get history: {e!s}")
          return OperationResult(
              success=False, message="Failed to get history", error=str(e)
          ).model_dump()

• src/csv_editor/models/csv_session.py:312-329 (handler)

  CSVSession.get_history - delegates to HistoryManager.get_history() and also returns statistics.

  def get_history(self, limit: int | None = None) -> dict[str, Any]:
      """Get operation history."""
      if not self.history_manager:
          # Return legacy history if new history is not enabled
          return {
              "success": True,
              "history": self.operations_history[-limit:] if limit else self.operations_history,
              "total": len(self.operations_history),
          }
  
      try:
          history = self.history_manager.get_history(limit)
          stats = self.history_manager.get_statistics()
  
          return {"success": True, "history": history, "statistics": stats}
      except Exception as e:
          logger.error(f"Error getting history: {e!s}")
          return {"success": False, "error": str(e)}

• src/csv_editor/models/history_manager.py:308-321 (handler)

  HistoryManager.get_history - the core implementation that builds a list of operation history dicts with index, is_current, and can_restore fields.

  def get_history(self, limit: int | None = None) -> list[dict[str, Any]]:
      """Get operation history."""
      history_list = []
  
      start = 0 if limit is None else max(0, len(self.history) - limit)
  
      for i, entry in enumerate(self.history[start:], start=start):
          history_dict = entry.to_dict()
          history_dict["index"] = i
          history_dict["is_current"] = i == self.current_index
          history_dict["can_restore"] = entry.data_snapshot is not None
          history_list.append(history_dict)
  
      return history_list

• src/csv_editor/models/history_manager.py:424-456 (helper)

  Helper that computes and returns statistics about the history (total operations, undo/redo state, operation type counts, etc.), called alongside get_history by CSVSession.get_history.

  def get_statistics(self) -> dict[str, Any]:
      """Get history statistics."""
      if not self.history:
          return {
              "total_operations": 0,
              "operation_types": {},
              "first_operation": None,
              "last_operation": None,
              "snapshots_count": 0,
          }
  
      # Count operation types
      operation_types = {}
      snapshots_count = 0
  
      for entry in self.history:
          operation_types[entry.operation_type] = operation_types.get(entry.operation_type, 0) + 1
          if entry.data_snapshot is not None:
              snapshots_count += 1
  
      return {
          "total_operations": len(self.history),
          "current_position": self.current_index + 1,
          "can_undo": self.can_undo(),
          "can_redo": self.can_redo(),
          "redo_stack_size": len(self.redo_stack),
          "operation_types": operation_types,
          "first_operation": self.history[0].timestamp.isoformat() if self.history else None,
          "last_operation": self.history[-1].timestamp.isoformat() if self.history else None,
          "snapshots_count": snapshots_count,
          "storage_type": self.storage_type.value,
          "max_history": self.max_history,
      }