Skip to content
48 changes: 48 additions & 0 deletions flower/api/tasks.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@

from ..utils import tasks
from ..utils.broker import Broker
from ..utils.tasks import parse_args, parse_kwargs, make_json_serializable
from . import BaseApiHandler

logger = logging.getLogger(__name__)
Expand Down Expand Up @@ -636,3 +637,50 @@ def get(self, taskid):
response['worker'] = task.worker.hostname

self.write(response)

class TaskReapply(BaseTaskHandler):
@web.authenticated
async def post(self, taskid):
"""
Get task info and reapply the task with the same arguments.

:param taskid: ID of the task to reapply.

Copilot AI Jan 7, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Missing API documentation: The TaskReapply endpoint lacks comprehensive API documentation that other endpoints in this file have (such as TaskInfo). Consider adding proper docstring documentation including HTTP method, example request/response, parameters description, and status codes.

Suggested change
Get task info and reapply the task with the same arguments.
:param taskid: ID of the task to reapply.
Reapply a previously executed task using the same arguments.
This endpoint retrieves an existing task by its ID, extracts its original
positional and keyword arguments, and submits a new task with the same
name and arguments. A new task ID is returned for the re-applied task.
**Example request**:
.. sourcecode:: http
POST /api/task/reapply/7b3b9f52-1af3-4e0c-8a8a-5a5f9c2f8c64 HTTP/1.1
Host: localhost
Accept: application/json
Cookie: user=...
**Example response**:
.. sourcecode:: http
HTTP/1.1 200 OK
Content-Type: application/json
{
"task-id": "c1a2b3d4-5678-90ab-cdef-1234567890ab",
"state": "PENDING"
}
:param str taskid: ID of the original task to reapply.
:statuscode 200: task successfully reapplied; returns new task ID and state (if backend configured)
:statuscode 400: invalid task arguments or original task has no name
:statuscode 401: unauthorized request
:statuscode 404: unknown task ID or unknown task name
:statuscode 500: internal error while reapplying the task

Copilot uses AI. Check for mistakes.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 3bc8db2. Added full API documentation in the same format as TaskInfo/TaskApply: example request/response and all status codes (200/400/401/404/500).

"""
# Get original task info
task = tasks.get_task_by_id(self.application.events, taskid)
Comment thread
yashpapa6969 marked this conversation as resolved.
if not task:
raise HTTPError(404, f"Unknown task '{taskid}'")

# Get task name
taskname = task.name
if not taskname:
raise HTTPError(400, "Cannot reapply task with no name")

try:
# Get the task object from registered tasks
task_obj = self.capp.tasks[taskname]
except KeyError as exc:
raise HTTPError(404, f"Unknown task '{taskname}'") from exc

# Parse args and kwargs from the original task
try:
args = parse_args(task.args)
kwargs = parse_kwargs(task.kwargs)
except Exception as exc:
Comment thread
auvipy marked this conversation as resolved.
Outdated
logger.error("Error parsing task arguments: %s", exc)
raise HTTPError(400, f"Invalid task arguments: {str(exc)}") from exc

# Apply the task with original arguments
try:
# Ensure args and kwargs are JSON serializable
args = make_json_serializable(args)
kwargs = make_json_serializable(kwargs)

result = task_obj.apply_async(args=args, kwargs=kwargs)
response = {'task-id': result.task_id}
if self.backend_configured(result):
response.update(state=result.state)
self.write(response)
except Exception as exc:
logger.error("Error reapplying task with args=%s, kwargs=%s: %s", args, kwargs, str(exc))
raise HTTPError(500, f"Error reapplying task: {str(exc)}") from exc
Comment thread
yashpapa6969 marked this conversation as resolved.
Outdated
33 changes: 33 additions & 0 deletions flower/static/js/flower.js
Original file line number Diff line number Diff line change
Expand Up @@ -690,4 +690,37 @@ var flower = (function () {

});

$('#task-retry').click(function () {
const $button = $(this);
const $spinner = $button.find('.spinner-border');
const taskId = $('#taskid').text();

if (!taskId) {
show_alert('Task ID is missing. Cannot proceed.', 'danger');
return;
}

// Show loading state
$button.prop('disabled', true);
$spinner.removeClass('d-none');

// Reapply the task using the reapply endpoint
$.ajax({
type: 'POST',
url: url_prefix() + '/api/task/reapply/' + taskId,
success: function (response) {
show_alert(`Task ${taskId} has been retried (new task ID: ${response['task-id']})`, 'success');
Comment thread
yashpapa6969 marked this conversation as resolved.
Outdated
// Optionally reload the page after success
setTimeout(() => location.reload(), 1500);
},
error: function (response) {
show_alert(response.responseText || 'Failed to retry task', 'danger');
Comment thread
yashpapa6969 marked this conversation as resolved.
Outdated
// Reset button state on error
$button.prop('disabled', false);
$spinner.addClass('d-none');
}
});
});


}(jQuery));
7 changes: 6 additions & 1 deletion flower/templates/task.html
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,11 @@ <h2>{{ getattr(task, 'name', None) }}
<button class="btn btn-danger float-end" id="task-terminate">Terminate</button>
{% elif task.state == "RECEIVED" or task.state == "RETRY" %}
<button class="btn btn-danger float-end" id="task-revoke">Revoke</button>
{% elif task.state == "FAILURE" %}
<button class="btn btn-warning float-end" id="task-retry" data-bs-toggle="button">
<span class="spinner-border spinner-border-sm d-none" role="status" aria-hidden="true"></span>
Retry
</button>
Comment thread
yashpapa6969 marked this conversation as resolved.
{% end %}
</h2>
</div>
Expand Down Expand Up @@ -89,4 +94,4 @@ <h2>{{ getattr(task, 'name', None) }}
</div>
</div>
</div>
{% end %}
{% end %}
1 change: 1 addition & 0 deletions flower/urls.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@
(r"/api/task/timeout/(.+)", control.TaskTimout),
(r"/api/task/rate-limit/(.+)", control.TaskRateLimit),
(r"/api/task/revoke/(.+)", control.TaskRevoke),
(r"/api/task/reapply/(.+)", tasks.TaskReapply),
# Metrics
(r"/metrics", monitor.Metrics),
(r"/healthcheck", monitor.Healthcheck),
Expand Down
52 changes: 52 additions & 0 deletions flower/utils/tasks.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
import datetime
import time
import json
import ast

from .search import parse_search_terms, satisfies_search_terms

Expand Down Expand Up @@ -68,3 +70,53 @@ def get_task_by_id(events, task_id):

def as_dict(task):
return task.as_dict()

def parse_args(args):
"""
Parse and process the `args` of the task.
"""
if not args:
return []
try:
# Attempt to parse JSON
parsed_args = json.loads(args)
if isinstance(parsed_args, str) and parsed_args.startswith('(') and parsed_args.endswith(')'):
return ast.literal_eval(parsed_args) # Handle stringified tuples safely
Comment thread
yashpapa6969 marked this conversation as resolved.
Outdated
return parsed_args
except (json.JSONDecodeError, SyntaxError):
# Fallback for stringified tuples or ellipsis
if args == '...':
return [...]
Comment thread
yashpapa6969 marked this conversation as resolved.
Outdated
if args.startswith('(') and args.endswith(')'):
return ast.literal_eval(args)
Comment thread
yashpapa6969 marked this conversation as resolved.
Outdated
return [args]

def parse_kwargs(kwargs):
"""
Parse and process the `kwargs` of the task.
"""
if not kwargs:
return {}
try:
# Attempt to parse JSON
return json.loads(kwargs)
except json.JSONDecodeError:
try:
# Fallback for stringified dictionaries
if kwargs.startswith('{') and kwargs.endswith('}'):
return ast.literal_eval(kwargs)
except (ValueError, SyntaxError):
return {}
Comment thread
yashpapa6969 marked this conversation as resolved.
Outdated
return {}
Comment thread
auvipy marked this conversation as resolved.
Outdated

def make_json_serializable(obj):
"""
Recursively replace non-serializable types with JSON-serializable alternatives.
"""
if isinstance(obj, list):
return [make_json_serializable(item) for item in obj]
elif isinstance(obj, dict):
return {key: make_json_serializable(value) for key, value in obj.items()}
Comment thread
yashpapa6969 marked this conversation as resolved.
Outdated
elif obj is Ellipsis:
return None # Replace `...` with `null`
return obj # Return the object if it's already serializable