Retrieve task results by id

[faroit]

Hi, first I want to thank you for this great package. I just started using it and it seems to really work well for me so far.

In my flask based application (built upon your flask example), I want use the results backend to offer an API endpoint that allows to return the results of a given job.

From the celery FAQ, it seems like you can retrieve the results given the unique task id:

result = MyTask.AsyncResult(task_id)
result.get()

Now, how would I implement such a functionality with dramatiq?

[Bogdanp]

I'm afraid this is a little janky right now, but say you have an actor called add, tied to a results backend:

@dramatiq.actor(store_results=True)
def add(x, y):
  return x + y

# you can give the user back a URI containing the message id
message = add.send(1, 2)
print(f"https://example.com/?message_id={message.message_id}")

# then when they visit that uri, you can construct a new message with that same id
# note that you have to construct the message against the same actor you used initially
message = add.message().copy(message_id=params["message_id"])
message.get_result()

# alternatively, you can pass the message to Backend.get_result
message = add.message().copy(message_id=params["message_id"])
backend.get_result(message)

Not ideal, but I hope that helps!

[faroit]

Thanks, that works. Before I close this, allow me a small follow up: How can I remove the results from the backend?

in celery it seems this can be done by:

result = async_result.get()
async_result.forget()

[Bogdanp]

There's currently no API to do this. You should be able to call backend.build_message_key(message) and then delete that key from memcached or redis for now.

[faroit]

indeed... result_backend.client.delete(result_backend.build_message_key(message)) works.

even though I have no experience with interfacing redis directly, it looks like accessing the client object directly is convenient enough. But maybe you could add these things to an FAQ or Wiki here.

[cmin764]

Indeed, this it will be very helpful to be added in the next release, because that's usually the main purpose for some simple task processing queues. To "futurize" a process and get the results by wire at client's convenience without going through any intermediate memcache/database storage (by just using dramatiq only).

A <task_function>.message_with_options(message_id=<previous_id>) would be enough.