restore_to_operation
Restore CSV Editor session data to a specific operation point using session and operation IDs, enabling precise data recovery and manipulation in large-scale CSV processing.
Instructions
Restore session data to a specific operation point.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| operation_id | Yes | ||
| session_id | Yes |
Implementation Reference
- src/csv_editor/server.py:525-532 (registration)MCP tool registration and handler wrapper for 'restore_to_operation', delegates to internal implementation.@mcp.tool async def restore_to_operation( session_id: str, operation_id: str, ctx: Context = None ) -> Dict[str, Any]: """Restore session data to a specific operation point.""" return await _restore_to_operation(session_id, operation_id, ctx)
- Internal handler implementation that retrieves the session and calls the session's restore_to_operation method, handling errors and logging.async def restore_to_operation( session_id: str, operation_id: str, ctx: Context = None ) -> Dict[str, Any]: """ Restore session data to a specific operation point. Args: session_id: Session identifier operation_id: Operation ID to restore to ctx: FastMCP context Returns: Dict with success status and restore result """ 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"Restoring session {session_id} to operation {operation_id}") result = await session.restore_to_operation(operation_id) if result["success"]: if ctx: await ctx.info(f"Successfully restored to operation {operation_id}") return OperationResult( success=True, message=result["message"], session_id=session_id, data=result ).model_dump() else: return OperationResult( success=False, message="Failed to restore to operation", error=result.get("error") ).model_dump() except Exception as e: logger.error(f"Error restoring to operation: {str(e)}") if ctx: await ctx.error(f"Failed to restore to operation: {str(e)}") return OperationResult( success=False, message="Failed to restore to operation", error=str(e) ).model_dump()
- CSVSession method that delegates to history_manager.restore_to_operation, updates the dataframe, and triggers auto-save if configured.async def restore_to_operation(self, operation_id: str) -> Dict[str, Any]: """Restore data to a specific operation point.""" if not self.history_manager: return {"success": False, "error": "History is not enabled"} try: data_snapshot = self.history_manager.restore_to_operation(operation_id) if data_snapshot is not None: self.df = data_snapshot # Trigger auto-save if configured if self.auto_save_manager.should_save_after_operation(): await self.auto_save_manager.trigger_save(self._save_callback, "restore") return { "success": True, "message": f"Restored to operation {operation_id}", "shape": self.df.shape } else: return { "success": False, "error": f"Could not restore to operation {operation_id}" } except Exception as e: logger.error(f"Error during restore: {str(e)}") return {"success": False, "error": str(e)}
- Core implementation in HistoryManager: locates the target operation, finds the nearest prior snapshot, updates current index, clears redo stack, saves state, and returns the snapshot DataFrame.def restore_to_operation(self, operation_id: str) -> Optional[pd.DataFrame]: """Restore data to a specific operation point.""" # Find the operation target_index = None for i, entry in enumerate(self.history): if entry.operation_id == operation_id: target_index = i break if target_index is None: logger.error(f"Operation {operation_id} not found") return None # Find the nearest snapshot at or before target snapshot = None for i in range(target_index, -1, -1): if self.history[i].data_snapshot is not None: snapshot = self.history[i].data_snapshot.copy() self.current_index = target_index # Clear redo stack since we're jumping to a specific point self.redo_stack.clear() # Save state if self.storage_type != HistoryStorage.MEMORY: self._save_history() logger.info(f"Restored to operation {operation_id}") return snapshot logger.error(f"No snapshot available for operation {operation_id}") return None