有没有办法使用 Flask Socket IO 登录用户?
Is there a way to login a user using Flask Socket IO?
我正在使用 Flask 与 Flask SocketIO 和 Flask Login.
我的问题是,当我尝试在 socketio.on
事件中使用 login_user
函数时,没有出现错误,但用户未登录。我的示例可能有助于说明这个。 my example event
将被调用以供用户登录,然后 /usersonly
将在该事件被调用后调用。
@socketio.on('my example event')
def my_func(json):
...
login_user(user)
@app.route('/usersonly')
@login.user_required
def my_call():
print(current_user)
如何才能让 my_func
可以登录用户?
使用 Flask Login 是不可能的,因为在文档中,它说 Flask Login 使用 Cookies 或 Header keys 进行身份验证,并且不可能在 WebSocket(Socket IO) 中访问它,所以我建议使用基于令牌的系统,就像在 API 中一样,使用 JWT 令牌。
@socket.on('login')
def on_login(message):
user = User.query.filter_by(email=message['email']).first()
if not user:
socket.emit('login', {'msg': "User not found"})
if not user.check_password(form['password']):
socket.emit('login', {'msg': "User not found"})
return
user = UserSchema().dump(user).data
token = jwt.encode(
{'id': user['id'], 'user': user['name']})
user['token'] = token.decode('UTF-8')
socket.emit('login', {'user': user})
然后为登录创建装饰器:
def login_required(f):
@wraps(f)
def decorated(message):
try:
data = jwt.decode(message['token'], globals.current_app.config['SECRET_KEY'])
try:
user = User.query.filter_by(id=data['id']).first()
Globals.current_user = UserSchema().dump(user).data
except Exception as e:
return Error.api_response(Error.USER_NOT_FOUND)
except Exception as e:
return Error.api_response(Error.INVALID_TOKEN)
return f(message)
return decorated
然后你可以像这样使用它:
@socketio.on('my example event')
@login_required
def my_func(json):
...
但请记住,您需要始终在 socketio 中传递令牌,以进行服务器验证,我建议阅读 Miguel Grinberh 的博客 rest auth in flask
我正在使用 Flask 与 Flask SocketIO 和 Flask Login.
我的问题是,当我尝试在 socketio.on
事件中使用 login_user
函数时,没有出现错误,但用户未登录。我的示例可能有助于说明这个。 my example event
将被调用以供用户登录,然后 /usersonly
将在该事件被调用后调用。
@socketio.on('my example event')
def my_func(json):
...
login_user(user)
@app.route('/usersonly')
@login.user_required
def my_call():
print(current_user)
如何才能让 my_func
可以登录用户?
使用 Flask Login 是不可能的,因为在文档中,它说 Flask Login 使用 Cookies 或 Header keys 进行身份验证,并且不可能在 WebSocket(Socket IO) 中访问它,所以我建议使用基于令牌的系统,就像在 API 中一样,使用 JWT 令牌。
@socket.on('login')
def on_login(message):
user = User.query.filter_by(email=message['email']).first()
if not user:
socket.emit('login', {'msg': "User not found"})
if not user.check_password(form['password']):
socket.emit('login', {'msg': "User not found"})
return
user = UserSchema().dump(user).data
token = jwt.encode(
{'id': user['id'], 'user': user['name']})
user['token'] = token.decode('UTF-8')
socket.emit('login', {'user': user})
然后为登录创建装饰器:
def login_required(f):
@wraps(f)
def decorated(message):
try:
data = jwt.decode(message['token'], globals.current_app.config['SECRET_KEY'])
try:
user = User.query.filter_by(id=data['id']).first()
Globals.current_user = UserSchema().dump(user).data
except Exception as e:
return Error.api_response(Error.USER_NOT_FOUND)
except Exception as e:
return Error.api_response(Error.INVALID_TOKEN)
return f(message)
return decorated
然后你可以像这样使用它:
@socketio.on('my example event')
@login_required
def my_func(json):
...
但请记住,您需要始终在 socketio 中传递令牌,以进行服务器验证,我建议阅读 Miguel Grinberh 的博客 rest auth in flask