129 lines
No EOL
3.9 KiB
Python
Executable file
129 lines
No EOL
3.9 KiB
Python
Executable file
"""
|
|
Task management routes for handling cancellable operations.
|
|
"""
|
|
|
|
from quart import Blueprint, jsonify, request
|
|
from app.services.task_manager import get_task_manager
|
|
from app.websocket_manager_async import get_async_websocket_manager
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
tasks_bp = Blueprint('tasks', __name__)
|
|
|
|
|
|
@tasks_bp.route('/<task_id>', methods=['DELETE'])
|
|
async def cancel_task(task_id: str):
|
|
"""
|
|
Cancel a running task by its ID.
|
|
|
|
Args:
|
|
task_id: The unique identifier of the task to cancel
|
|
|
|
Returns:
|
|
JSON response indicating success or failure
|
|
"""
|
|
try:
|
|
task_manager = get_task_manager()
|
|
|
|
# Get task info before cancellation for WebSocket notification
|
|
task_info = await task_manager.get_task_info(task_id)
|
|
|
|
# Attempt to cancel the task
|
|
cancelled = await task_manager.cancel_task(task_id)
|
|
|
|
if not cancelled:
|
|
return jsonify({
|
|
'error': 'Task not found or already completed',
|
|
'task_id': task_id
|
|
}), 404
|
|
|
|
# Send WebSocket notification about cancellation
|
|
websocket_manager = get_async_websocket_manager()
|
|
if task_info and task_info.user_id:
|
|
await websocket_manager.emit_to_user(
|
|
task_info.user_id,
|
|
'task_cancelled',
|
|
{
|
|
'task_id': task_id,
|
|
'task_type': task_info.task_type,
|
|
'message': f'{task_info.task_type.replace("_", " ").title()} cancelled successfully'
|
|
}
|
|
)
|
|
|
|
logger.info(f"Successfully cancelled task {task_id}")
|
|
|
|
return jsonify({
|
|
'message': 'Task cancelled successfully',
|
|
'task_id': task_id,
|
|
'task_type': task_info.task_type if task_info else None
|
|
}), 200
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error cancelling task {task_id}: {str(e)}")
|
|
return jsonify({
|
|
'error': 'Internal server error while cancelling task',
|
|
'task_id': task_id
|
|
}), 500
|
|
|
|
|
|
@tasks_bp.route('/user/<user_id>', methods=['GET'])
|
|
async def get_user_tasks(user_id: str):
|
|
"""
|
|
Get all active tasks for a specific user.
|
|
|
|
Args:
|
|
user_id: The ID of the user whose tasks to retrieve
|
|
|
|
Returns:
|
|
JSON response with list of active tasks
|
|
"""
|
|
try:
|
|
task_manager = get_task_manager()
|
|
user_tasks = await task_manager.get_user_tasks(user_id)
|
|
|
|
# Convert task info to JSON-serializable format
|
|
tasks_data = []
|
|
for task_id, task_info in user_tasks.items():
|
|
tasks_data.append({
|
|
'task_id': task_id,
|
|
'task_type': task_info.task_type,
|
|
'status': task_info.status,
|
|
'created_at': task_info.created_at.isoformat(),
|
|
'metadata': task_info.metadata
|
|
})
|
|
|
|
return jsonify({
|
|
'tasks': tasks_data,
|
|
'count': len(tasks_data)
|
|
}), 200
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error fetching tasks for user {user_id}: {str(e)}")
|
|
return jsonify({
|
|
'error': 'Internal server error while fetching user tasks'
|
|
}), 500
|
|
|
|
|
|
@tasks_bp.route('/status', methods=['GET'])
|
|
async def get_task_status():
|
|
"""
|
|
Get overall task manager status (for debugging/monitoring).
|
|
|
|
Returns:
|
|
JSON response with task manager statistics
|
|
"""
|
|
try:
|
|
task_manager = get_task_manager()
|
|
active_count = await task_manager.get_active_task_count()
|
|
|
|
return jsonify({
|
|
'active_tasks': active_count,
|
|
'status': 'operational'
|
|
}), 200
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error fetching task manager status: {str(e)}")
|
|
return jsonify({
|
|
'error': 'Internal server error while fetching status'
|
|
}), 500 |