Flask 实现 Rest API (03) - 用户登陆

Web 服务是无状态的，那么对于客户端提交的请求，服务器如何判断请求的合法性呢？比如只能登陆成功的用户，服务器才按要求返回数据。本文提供一种利用会话 (session) 机制实现保存用户登录信息的方法。

利用 session 记录登录信息

为了在服务器端保存用户的登录信息，需要在服务器端保存会话数据，浏览器每次请求的时候，都需要携带会话数据，服务器端进行校验。这种以用户鉴权为目的的 session 也可也被称为 token。大致的流程如下：

客户端使用用户名和密码请求登录
服务端收到请求，验证用户名与密码
验证成功后，服务器端会签发一个 Token，再把这个 Token 发送给客户端
客户端收到 Token 以后可以把它存储起来，比如放在 Cookie 里或者 Local Storage 里
客户端每次向服务器端请求资源的时候需要带着服务端签发的 Token
服务器端收到请求，去验证客户端请求里面携带的 Token，如果验证成功，就向客户端返回请求的数据

Flask 实现 session

Flask 实现 session，对每个客户端的会话分配一个会话 ID。会话数据存储在 Headers 的 cookie 中，服务器以加密方式签名。对于这种加密，Flask application 需要定义一个 SECRET_KEY ，这是一个字符串，可以理解为秘钥，可以自由设置，尽量复杂为好。

app.config['SECRET_KEY'] = 'your_secret_key_here'

然后可以设置会话对象（字典类型的对象），比如在服务器设置 username 的会话对象：

@main.route('/token', methods=['POST'])
def login():
    username = request.json.get('username')
    password = request.json.get('password')
    remember_user = request.json.get('remember_user')

    if verify_user(username, password) == True:
        session['username'] = username
        if remember_user == True:
            session.permanent = True
        else:
            session.permanent = False

        return jsonify({'result': 1, 'description': 'Login successful.'}), 200
    else:
        return jsonify({'result': 0, 'description': 'Login failed.'}), 401

设置 session 对象 的代码语句 session['username'] = username。
会话对象是一个字典对象。要删除会话变量，使用 pop() 方法。

@main.route('/token', methods=['DELETE'])
def logout():
    if 'username' in session:
        session.pop('username')
        return jsonify({'result': 1,'description': 'Logout successful.'})
    else:
        return jsonify({'result': 0, 'description': 'No user was found.'})

设置 session 过期时间

登录之后，服务器和你的浏览器之间建立了一个 session，它通常有一个过期时间，比如 30 分钟，假设登录后 10 分钟你进行了一次操作，发出 get 请求，那么这个过期时间就要重新计算。从你操作的这一刻起，30 分钟以后过期。这就意味着，如果你 30 分钟内没有任何操作，session 就会过期，必须重新登录。

Flask 设置 session 的过期时间需要两条语句：

app.permanent_session_lifetime = datetime.timedelta(seconds=30*60)
session.permanent = True

这样就设置了 session 的过期时间为 30 分钟。

完整代码

# token.py

# encoding: utf-8

from .. import db

def get_users():
    conn = db.connect()
    curr = conn.cursor()

    curr.execute('select username, pwd as password from users')

    users_dict = [dict((curr.description[i][0], value)
        for i, value in enumerate(rows))
        for rows in curr.fetchall()]

    curr.close()
    conn.close()

    return users_dict


def get_password(user_name):
    result = ''
    users = get_users()
    for user in users:
        if user['username'].upper() == user_name.upper():
            result = user.get('password')

    return result


def verify_user(user_name, password):
    if password == get_password(user_name):
        return True
    else:
        return False

# token_routes.py

# encoding: utf-8

from . token import *
from . import main
from flask import request, session, jsonify

@main.route('/token', methods=['POST'])
def login():
    username = request.json.get('username')
    password = request.json.get('password')
    remember_user = request.json.get('remember_user')

    if verify_user(username, password) == True:
        session['username'] = username
        if remember_user == True:
            session.permanent = True
        else:
            session.permanent = False

        return jsonify({'result': 1, 'description': 'Login successful.'}), 200
    else:
        return jsonify({'result': 0, 'description': 'Login failed.'}), 401


@main.route('/token', methods=['DELETE'])
def logout():
    if 'username' in session:
        session.pop('username')
        return jsonify({'result': 1,'description': 'Logout successful.'})
    else:
        return jsonify({'result': 0, 'description': 'No user was found.'})

路由中添加 session 校验

在服务器端，对相关的视图函数 (view function）进行校验：

@main.route('/employees')
def listEmployees():
    if 'username' in session:
        emp = Employee();
        employees = emp.listAll()
        return jsonify({'rows': employees}), 200;
    else:
        return jsonify({'error': 'No authorization.'}), 401

客户端测试

使用自己喜欢的 Rest 测试工具，比如 Postman。本文提供利用 requests 库进行请求的示例：

import requests
import json


def get_session_cookie():
    result = ''

    payload = {
        "username": "admin",
        "password": "pwd"
    }

    resp = requests.post('http://localhost:5000/token', json=payload)

    for k, v in resp.cookies.items():
        if k == 'session':
            result = v

    return result

cookie = get_session_cookie()

def get_employees():
    session_cookie = {
        'session': cookie
    }
    resp = requests.get('http://localhost:5000/employees', cookies=session_cookie)

    return resp


def create_employee():
    session_cookie = {
        'session': cookie
    }

    payload = {
        "AGE": 26,
        "EDUCATION": "Lower secondary",
        "EMAIL": "p.harris@randatmail.com",
        "EMP_ID": 9001,
        "GENDER": "Female",
        "MARITAL_STAT": "Married",
        "NR_OF_CHILDREN": 4,
        "PHONE_NR": "980-0639-38"
    }

    resp = requests.post(
        'http://localhost:5000/employees/create',
        json=payload,
        cookies=session_cookie
    )

    return resp


def modify_employee():
    session_cookie = {
        'session': cookie
    }

    payload = {
        "AGE": 26,
        "EDUCATION": "Lower secondary",
        "EMAIL": "p.harris@randatmail.com",
        "EMP_ID": 9001,
        "GENDER": "女",
        "MARITAL_STAT": "Married",
        "NR_OF_CHILDREN": 4,
        "PHONE_NR": "980-0639-38"
    }

    resp = requests.put(
        'http://localhost:5000/employees/9001',
        json=payload,
        cookies=session_cookie)

    return resp


def delete_employee():
    session_cookie = {
        'session': cookie
    }

    resp = requests.delete(
        'http://localhost:5000/employees/9001',
        cookies=session_cookie)

    return resp

参考

Flask Sessions会话

小蜜蜂