from flask import Flask
from flask_admin import Admin
from flask_login import LoginManager
app = Flask(__name__)
admin = Admin(app)
# 1:实例化login_manager
login_manager = LoginManager()
# 2:设置密钥
app.config['SECRET_KEY'] = '234567890'
# 3:关联app
login_manager.init_app(app)
app.run()
你需要提供一个用户的回调函数。这个回调函数被用来从session中获取UserID,根据这个UserID获取User对象。
@login_manager.user_loader
def load_user(user_id):
return User.get(user_id)
注意: 如果这个USERID不存在, 你需要返回一个None, 不要抛出异常。
@app.route('/login', methods=['GET', 'POST'])
def login():
# 获取表单
form = LoginForm()
# 如果校验成功
if form.validate_on_submit():
# 进行登录
login_user(user)
# 刷新
flask.flash('Logged in successfully.')
# 获取下一个链接
next = flask.request.args.get('next')
# 判断链接安全性
if not is_safe_url(next):
return flask.abort(400)
# 重定向到下个页面。
return flask.redirect(next or flask.url_for('index'))
return flask.render_template('login.html', form=form)
模板中也可以使用该用户:
{% if current_user.is_authenticated %}
Hi {{ current_user.name }}!
{% endif %}
用户需要登录的视图,采用装饰器登录:
@app.route("/settings")
@login_required
def settings():
pass
用户注销时:
@app.route("/logout")
@login_required
def logout():
logout_user()
return redirect(somewhere)
1: 当用户没有登录的时候访问视图,Flask_login会发送一个消息并将他们重定向到视图。
login_manager.login_view = "users.login"
2: 我们也可以自定义消息内容:
login_manager.login_message = u"Bonvolu ensaluti por uzi tiun paĝon."
3: 自定义消息的级别:
login_manager.login_message_category = "info"
4: 如果想要自定义:
@login_manager.unauthorized_handler
def unauthorized():
# do stuff
return a_response
request_loader
回调。此回调的行为应与您的user_loader
回调相同,只是它接受 Flask 请求而不是 user_id。@login_manager.request_loader
def load_user_from_request(request):
# 获取请求的键
api_key = request.args.get('api_key')
# 如果有获取该用户,存在该用户则返回。
if api_key:
user = User.query.filter_by(api_key=api_key).first()
if user:
return user
# 获取认证密钥
api_key = request.headers.get('Authorization')
if api_key:
api_key = api_key.replace('Basic ', '', 1)
try:
# 解码
api_key = base64.b64decode(api_key)
except TypeError:
pass
# 根据密钥获取用户
user = User.query.filter_by(api_key=api_key).first()
if user:
return user
return None
默认情况下,当用户未实际登录时,current_user
设置为一个AnonymousUserMixin
对象。
如果您对匿名用户有自定义要求(例如,他们需要有一个权限字段),您可以提供一个可调用的(类或工厂函数)来创建匿名用户。
login_manager.anonymous_user = MyAnonymousUser
首次登录的时候,如果认证成功,首先会将用户id, Sessionid等信息存储到Session中。并将用户对象存入request.context中。
login_user源码解析:
def login_user(user, remember=False, force=False, fresh=True):
if not force and not user.is_active:
return False
# 从user对象中获取用户的ID
user_id = getattr(user, current_app.login_manager.id_attribute)()
# 将用户信息写入session中
session['user_id'] = user_id
session['_fresh'] = fresh
session['_id'] = current_app.login_manager._session_identifier_generator()
if remember:
session['remember'] = 'set'
# 将user对象存储进当前的request context
_request_ctx_stack.top.user = user
# 通过send来发射此signal,当注册监听此signal的回调函数收到此signal之后就会执行函数。
user_logged_in.send(current_app._get_current_object(), user=_get_user())
return True
非第一次登录: 首先携带session信息来请求,先判断是否在session中能够拿到用户ID, 如果能拿到,根据ID看看是否能查询到用户,如果能则授权成功,返回处理后的response对象。
视图登录保护:
@app.route('/')
@app.route('/main')
@login_required
def main():
return render_template(
'main.html', username=current_user.username)
分析@login_required:
# flask_login/utils.py
def login_required(func):
@wraps(func)
def decorated_view(*args, **kwargs):
# 如果request method为例外method,即在EXEMPT_METHODS中的method,可以不必鉴权
if request.method in EXEMPT_METHODS:
return func(*args, **kwargs)
# 如果_login_disabled为True则不必鉴权
elif current_app.login_manager._login_disabled:
return func(*args, **kwargs)
# 正常鉴权
elif not current_user.is_authenticated:
return current_app.login_manager.unauthorized()
return func(*args, **kwargs)
return decorated_view
默认情况下只有OPTIONS method在EXEMPT_METHODS set中,而GET、PUT、POST等常见的methods都需要鉴权。
_login_disabled
默认为False
正常鉴权的关键在于current_user.is_authenticated
是否为True,为True则正常处理请求,为False则进入unauthorized处理流程。那么这个current_user到底怎么就能鉴权了?它是怎么来的呢?来看下定义:
current_user = LocalProxy(lambda: _get_user())
原来current_user是一个LocalProxy对象,其代理的对象需要通过_get_user()
来获取,简单来说_get_user()会返回两种用户,一种是正常的用户对象(鉴权成功),一种是anonymous用户对象(鉴权失败)。而正常的用户对象其is_authenticated
属性总是为True,相对的anonymous用户对象的is_authenticated
属性总是为False。
分析: _get_user():
# flask_login/utils.py
def _get_user():
if has_request_context() and not hasattr(_request_ctx_stack.top, 'user'):
current_app.login_manager._load_user()
return getattr(_request_ctx_stack.top, 'user', None)
_get_user
函数时就会直接从request context中获取user对象return getattr(_request_ctx_stack.top, 'user', None)
但如果是非首次登陆,当前request context中并没有保存user对象,就需要调用current_app.login_manager._load_user()
来去load user对象。分析: _load_user()
# flask_login/login_manager.py
def _load_user(self):
user_accessed.send(current_app._get_current_object())
config = current_app.config
if config.get('SESSION_PROTECTION', self.session_protection):
deleted = self._session_protection()
if deleted:
return self.reload_user()
is_missing_user_id = 'user_id' not in session
if is_missing_user_id:
cookie_name = config.get('REMEMBER_COOKIE_NAME', COOKIE_NAME)
header_name = config.get('AUTH_HEADER_NAME', AUTH_HEADER_NAME)
has_cookie = (cookie_name in request.cookies and
session.get('remember') != 'clear')
if has_cookie:
return self._load_from_cookie(request.cookies[cookie_name])
elif self.request_callback:
return self._load_from_request(request)
elif header_name in request.headers:
return self._load_from_header(request.headers[header_name])
return self.reload_user()
_load_user
大体的过程是首先检查SESSION_PROTECTION设置,如果SESSION_PROTECTION 为strong或者basic类型,那么就会执行_session_protection()
动作,否则不执行此操作。_session_protection
在session_id不一致的时候(比如IP变化会导致session id的变化)才真正有用,这时,如果为basic类型或者session permanent为True时,只标注session为非新鲜的(not fresh);而如果为strong,则会删除session中的用户信息,并重新load user,即调用reload_user
。user_id
来判断),如果有则直接reload_user
,如果没有,则有三种方式来load user,一种是通过remember cookie,一种通过request,一种是通过request header,依次尝试。分析: reload_user():
# flask_login/login_manager.py
def reload_user(self, user=None):
ctx = _request_ctx_stack.top
if user is None:
user_id = session.get('user_id')
if user_id is None:
# 当无法获取到有效的用户id时,就认为是anonymous user
ctx.user = self.anonymous_user()
else:
# user callback就是我们通过@login_manager.user_loader装饰的函数,用于获取user object
if self.user_callback is None:
raise Exception(
"No user_loader has been installed for this "
"LoginManager. Add one with the "
"'LoginManager.user_loader' decorator.")
user = self.user_callback(user_id)
if user is None:
ctx.user = self.anonymous_user()
else:
ctx.user = user
else:
ctx.user = user