利用自定义 ORM 下使用 flask-

2023-01-30 23:31

短信预约 -IT技能 免费直播动态提醒

1. 安装；

pip install flask_login

2. 使用：

注册应用

import os
from flask_login import LoginManager, current_user

login_manager = LoginManager()
login_manager.login_view = 'users.login'  # 未登录作的跳转视图
login_manager.session_protection = 'strong'  
login_manager.login_message = u"Bonvolu ensaluti por uzi tiun paĝon."  # 设置快闪消息，用于提示用户

app = Flask(__name__) app.secret_key = os.urandom(24) login_manager.init_app(app)

自定义的 User 类需要提供一下属性：
- is_authenticated : 用来判断是否是已经授权了，如果通过授权就会返回true
- is_active : 判断是否已经激活,可用
- is_anonymous ：判断是否是匿名用户
- get_id() : 返回用户的唯一标识

class UserInfo(Model):
    __tablename__ = "report_user"
    id = Integer('id')
    user_id = BigInt('user_id', primary_key=True)
    union_id = BigInt('union_id')
    name = String('name')
    status = Integer('status', default=1)
    last_login_time = String('last_login_time', default=datetime.now)
    last_update_by = BigInt('last_update_by')
    create_time = Datetime('create_time', default="datetime.now")

    @property
    def is_authenticated(self):
        return self.status == 1

    @property
    def is_anonymous(self):
        return self.status == 2

    @property
    def is_active(self):
        if self.status is None:
            self.rQuery.Select("status").Where('chart_id=%s', [self.group_id])
        return self.status != 0

    def get_id(self):
        return self.user_id

3. 以 cookie 和 session 方式登录

配置：通过 session 中的信息导入用户实例

@login_manager.user_loader
def load_user(user_id):
    print("user_id =", user_id)
    msg = UserInfo.load_by_user_id(user_id)
    if user:
        user = UserInfo.to_model(msg)
    else:
        print("no found =", msg)
　　　　 user = msg
    return user

登录与退出
- 登录：通过 login_user(user), 把用户信息设置到 session 中
- 登出：通过 logout_user(), 清除此保存在缓冲中的cookie 和 session 中的信息

4. 自定义以请求信息头部或者传递的参数作为键值

　　这种情况下，使用 request_loader 回调，与 user_loader 回调原理一样，但是它不是接受user_id

@login_manager.request_loader
def load_user_from_request_by_args(request):

　　# 使用 api 参数，或者 post 携带的键值
　　union_id = request.args.get('union_id')
　　if union_id:
　　user = User.query.filter_by(union_id=union_id).first()
　　return user

@login_manager.request_loader
def load_user_from_request_by_headers(request):

　　# 使用 请求头部 headers 携带的键值
　　field = request.headers.get('Authorization')
　　if field:
　　    union_id = field.replace('Basic ', '', 1)
　　try:
　　    union_id = base64.b64decode(union_id)
　　except TypeError:
　　    pass
　　user = User.query.filter_by(union_id=union_id).first()
　　return user