Hey!
I need to run asynchronous functions in my Python + Flask application for a Slack app, which requires a response within 3 seconds while allowing longer processing.
I tried everything that Vercel’s AI suggested + other things I found online like:
- Using from concurrent.futures import ThreadPoolExecutor
- Using import threading
- Using Flask-Executor
The last one gave me this error:
RuntimeWarning: coroutine 'test_async' was never awaited
executor.submit(test_async(command))
And a mix of async python packages, in all of them the function ends as soon as I return on the request.
I saw that there is a waitUntil function in JS, but I didn’t find anything like that for python.
This is the code I’m testing:
@app.route('/slack/commands', methods=['POST'])
async def handle_commands():
if not signature_verifier.is_valid_request(request.get_data().decode('utf-8'), request.headers):
return jsonify({'status': 'invalid_request'}), 403
# Acknowledge Slack's command immediately to avoid timeouts
command = request.form
executor.submit(test_async(command))
# Return an immediate acknowledgment to Slack to avoid timeout
return jsonify("We are finding your stuff, give us a moment :grimacing:"), 200
async def test_async(command):
response_url = command.get('response_url')
try:
requests.post(response_url, json={"text": "Hello, World!", "response_type": "in_channel",})
# Simulate some long-running task
await asyncio.sleep(10)
requests.post(response_url, json={"text": "Waiting for 10 seconds!", "response_type": "in_channel",})
except Exception as e:
print(f"Error sending message to response_url: {e}")
(this is a test workflow to test this before we add our core functionality (that takes around 10 seconds) here.
Thanks for your time!