[Python] gevent AssertionError when using pymongo in flask-socketio app

• gevent version: 24.2.1 (pip install gevent)
• Python version: Python 3.12.3 (python --version)
• Operating System:

No LSB modules are available.
Distributor ID: Ubuntu
Description: Ubuntu 24.04.1 LTS
Release: 24.04
Codename: noble

Description:


Problem: When I run my flask socketio app with gunicorn it shows the following error.

[2024-09-19 19:40:20 +0530] [7874] [INFO] Starting gunicorn 23.0.0
[2024-09-19 19:40:20 +0530] [7874] [INFO] Listening at: http://0.0.0.0:8000 (7874)
[2024-09-19 19:40:20 +0530] [7874] [INFO] Using worker: geventwebsocket.gunicorn.workers.GeventWebSocketWorker
[2024-09-19 19:40:20 +0530] [7893] [INFO] Booting worker with pid: 7893
Traceback (most recent call last):
File "src/gevent/_abstract_linkable.py", line 287, in gevent._gevent_c_abstract_linkable.AbstractLinkable._notify_links
File "src/gevent/_abstract_linkable.py", line 333, in gevent._gevent_c_abstract_linkable.AbstractLinkable._notify_links
AssertionError: (None, <callback at 0x769b75efca40 args=([],)>)
2024-09-19T14:10:20Z <callback at 0x769b75efca40 args=([],)> failed with AssertionError


Command used to run: gunicorn -c gunicorn.conf.py app:app Gunicorn configuration:

workers = 1
worker_class = 'geventwebsocket.gunicorn.workers.GeventWebSocketWorker'
bind = '0.0.0.0:8000'
worker_connections = 1000
loglevel = 'info'
accesslog = '-'
errorlog = '-'
timeout = 120
preload_app = True

What I've run:


app.py:

from gevent import monkey
_ = monkey.patch_all()

import os
import jwt
from functools import wraps
from flask import Flask, render_template, request, redirect, make_response, url_for, send_from_directory
from utils.data_utils import extract_data as extract_data1688, get_driver as get_driver1688, LOAD_DELAY as LOAD_DELAY1688
from flask_socketio import SocketIO, emit
import random
import traceback
from pymongo.mongo_client import MongoClient
from pymongo.server_api import ServerApi
from dotenv import load_dotenv
load_dotenv()

app = Flask(__name__)
app.config['SECRET_KEY'] = os.getenv("JWT_SECRET")
socketio = SocketIO(app)
client = MongoClient("<exactly_correct_uri>", server_api=ServerApi('1'))

JWT_SECRET = os.getenv("JWT_SECRET")

def token_required(fn):
@wraps(fn)
def wrapper(*args, **kwargs):
token = request.cookies.get('auth_token')
if not token:
return redirect('/login')
try:
decoded_token = jwt.decode(token, JWT_SECRET, algorithms=['HS256'])
if decoded_token.get('role') != 'MOD':
return redirect('/login')
except jwt.ExpiredSignatureError:
return redirect('/login')
except jwt.InvalidTokenError:
return redirect('/login')
return fn(*args, **kwargs)
return wrapper

@app.route('/')
@token_required
def index():
return render_template('index.html')

@app.route('/login', methods=['GET', 'POST'])
def login():
message = request.args.get('message', "")
if request.method == 'POST':
token = request.form.get('auth_token')
try:
decoded_token = jwt.decode(token, JWT_SECRET, algorithms=['HS256'])
if decoded_token.get('role') != 'MOD':
return redirect(url_for('login', message="Unauthorized role."))
except jwt.ExpiredSignatureError:
return redirect(url_for('login', message="Token expired. Please try again."))
except jwt.InvalidTokenError:
return redirect(url_for('login', message="Invalid token. Please try again."))

resp = make_response(redirect('/'))
resp.set_cookie('auth_token', token)
return resp

return render_template('login.html', message=message)

@app.route('/public/<path:filename>')
def serve_public_file(filename):
return send_from_directory('public', filename)

@socketio.on("data_scrape_start")
@token_required
def data_scrape_start(url):
print("[scraper] Scraping:", url)
driver = None
try:
emit("log", {"message": f"Scraping: {url}", "level": "i"})
socketio.sleep(random.uniform(2, 5))
driver = get_driver1688(True)
driver.get(url)
socketio.sleep(LOAD_DELAY1688)
data = extract_data1688(driver, url, socketio)
emit("data", data)
emit("log", {"message": "Fetched all data!", "level": "s"})
emit("log", {"message": "Uploading to database...", "level": "i"})
# collection.insert_one(data)
emit("log", {"message": "Uploaded all data!", "level": "s"})
except Exception as e:
emit("log", {"message": f"Error: {str(e)}", "level": "e"})
traceback.print_exc()
finally:
emit("data_scrape_end")
if driver:
driver.quit()


if __name__ == '__main__':
socketio.run(app, host='0.0.0.0', port=8000, debug=True)


Extra: When i run the application by simply using python app.py I do not see any error and it works fine i guess. Also the error only shows up when i include database code, otherwise it doesn't.

Continue reading...