唐抉的个人博客

Flask实战之搭建博客后端

Flask Web 后端 Web框架
字数统计: 5.7k阅读时长: 30 min
2022/10/24

创建一个Flask RESTful API

本节的目的为搭建Flask应用，并提供一个测试API，客户端访问/ping后会返回pong!响应。

配置Flask

确保python3已安装后，在合适的位置新建tutorproject项目目录，在tutorproject里新建back-end目录，作为后端API应用所在的位置。在C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end的目录下右键，选择在终端打开后，按照以下步骤搭建Flask所需的环境。

1
2
3
4
5
6
7
8
#创建虚拟环境venv
PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end>python -m venv venv
#激活环境venv
PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end>venv\Scripts\activate
#虚拟环境中导入flask模块
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end>pip install flask
#把系统环境信息写到txt中
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end>pip freeze >requirements.txt

应用工厂

创建tutorweb目录，并在tutorweb目录下新建__init__.py文件，在__init__.py中，使用应用工厂函数来创建Flask应用：

1
2
3
4
5
6
7
8
9
10
11
12
from flask import Flask
from config import Config

def create_app(config_class=Config):
    app=Flask(__name__)
    app.config.from_object(config_class)

    #注册blueprint
    from tutorweb.api import bp as api_bp
    app.register_blueprint(api_bp,url_prefix='/api')
    
    return app

创建tutorweb/api目录，并在api目录下新建__init__.py文件，定义蓝图：

1
2
3
4
from flask import Blueprint
bp=Blueprint('api',__name__)
#防止循环导入ping.py文件
from tutorweb.api import ping

在api目录下新建ping.py文件，定义路由函数，当客户端访问/ping时返回包含JSON的数据：

1
2
3
4
5
6
7
from flask import jsonify
from tutorweb.api import bp

@bp.route('/ping',methods=['GET'])
def ping():
    #vue.js用来测试与后端Flask API的连通性
    return jsonify('Pong!')

应用启动文件

在back-end目录下新建madblog.py文件：

1
2
3
from tutorweb import create_app

app=create_app()

配置文件

在back-end目录下新建config.py文件：

1
2
3
4
5
6
7
8
import os 
from dotenv import load_dotenv

basedir=os.path.abspath(os.path.dirname(__file__))
load_dotenv(os.path.join(basedir,'.env'),encoding='utf-8')

class Config(object):
    pass

读取环境变量信息

将Flask应用所需的系统环境变量写到back-end/.env中，可先使用python-dotenv这个包来读取环境变量信息，再把系统环境写到txt中：

1
2
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end>pip install python-dotenv
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end>pip freeze >requirements.txt

在back-end目录下新建.env文件，用记事本打开后写入以下信息保存：

1
2
FLASK_APP=madblog.py
FLASK_DEBUG=1

启动应用

输入flask run即可启动应用：

1
2
3
4
5
6
7
8
9
10
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end>flask run 

 * Serving Flask app 'madblog.py'
 * Debug mode: on
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on http://127.0.0.1:5000
Press CTRL+C to quit
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 354-979-609

启动应用后，打开浏览器，访问http://127.0.0.1:5000/api/ping，若出现Pong!则说明所定义的Ping-Pong测试路由正常。

在cmd页面按下ctrl+c即可停止应用运行。

此时的目录结构如下：

Flask设计User用户相关API

本节的内容为：Flask后端针对“用户资源”提供部分RESTful API，基于token认证，支持添加用户、查看单个或多个用户、修改用户，使用HTTPie或Postman测试API通过。

数据库

ORM：SQLAlchemy

安装Flask-SQLAlchemy插件和数据表结构有变化后进行迁移的Flask-Migrate插件：

1
2
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end>pip install flask-sqlalchemy flask-migrate
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end>pip freeze > requirements.txt

修改配置文件back-end/config.py，默认使用SQLite数据库：

1
2
3
4
5
6
7
8
9
10
import os 
from dotenv import load_dotenv

basedir=os.path.abspath(os.path.dirname(__file__))
load_dotenv(os.path.join(basedir,'.env'),encoding='utf-8')

class Config(object):
    SQLALCHEMY_DATABASE_URI=os.environ.get('DATABASE_URL') or \
        'sqlite:///'+os.path.join(basedir,'tutorweb.db')
    SQLALCHEMY_TRACK_MODIFICATIONS=False

修改tutorweb/__init__.py，引入并初始化插件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from flask import Flask
from flask_cors import CORS
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from config import Config

#定义Flack_SQLAlchemy
db=SQLAlchemy()

#定义Flack_Migrate 
migrate=Migrate()

def create_app(config_class=Config):
    app=Flask(__name__)
    app.config.from_object(config_class)

    #启用CORS
    CORS(app)
    #初始化Flack_SQLAlchemy
    db.init_app(app)
    #初始化Flack_Migrate 
    migrate.init_app(app,db)

    #注册blueprint
    from tutorweb.api import bp as api_bp
    app.register_blueprint(api_bp,url_prefix='/api')
    
    return app

在back-end目录下新建tutorweb.db文件作为数据库文件，修改back-end目录下的madblog.py文件如下：

1
2
3
from tutorweb import create_app,db

app=create_app()

定义User用户数据模型

创建tutorweb/models.py:

1
2
3
4
5
6
7
8
9
10
from tutorweb import db

class User(db.Model):
    id=db.Column(db.Integer,primary_key=True)
    username=db.Column(db.String(64),index=True,unique=True)
    email=db.Column(db.String(120),index=True,unique=True)
    password_hash=db.Column(db.String(128))#不保留原始密码

    def __repr__(self):
        return '<User {}>'.format(self.username)

修改tutorweb/__init__.py，在文件末尾添加：

1
from tutorweb import models

第一次数据库迁移

创建迁移数据库：

1
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end>flask db init

生成迁移脚本：

1
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end>flask db migrate -m "add users table"

将迁移脚本应用到数据库中：

1
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end>flask db upgrade

若要回滚上次的迁移，可用flask db downgrade命令回滚。

存储用户密码的hash值：

使用werkzeug.security库的generate_password_hash和check_password_hash来创建哈希密码和验证密码的hash是否一致：

1
2
3
4
5
6
7
8
9
10
11
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end> python
Python 3.10.5 (tags/v3.10.5:f377153, Jun  6 2022, 16:14:13) [MSC v.1929 64 bit (AMD64)] on win32
Type "help", "copyright", "credits" or "license" for more information.
>>> from werkzeug.security import generate_password_hash, check_password_hash
>>> hash = generate_password_hash('123')
>>> hash
'pbkdf2:sha256:260000$RBvmqvkVP61YXGy0$52d3ea1ae76f51d284ba66d2665c8bcfdd069b47cdff7e3211e46d5d98a9d01a'
>>> check_password_hash(hash, 'foobar')
False
>>> check_password_hash(hash, '123')
True

注意：使用generate_password_hash生成不同的3个'123'哈希值，并复制其值保存下来备用，以作为后面用户密码使用。

修改tutorweb/models.py，增加创建密码和验证密码两个方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from werkzeug.security import generate_password_hash,check_password_hash
from tutorweb import db

class User(db.Model):
    id=db.Column(db.Integer,primary_key=True)
    username=db.Column(db.String(64),index=True,unique=True)
    email=db.Column(db.String(120),index=True,unique=True)
    password_hash=db.Column(db.String(128))#不保留原始密码

    def __repr__(self):
        return '<User {}>'.format(self.username)

    def set_password(self,password):
        self.password_hash=generate_password_hash(password) 

    def check_password(self,password):
        return check_password_hash(self.password_hash,password)

配置Flask Shell上下文环境：

flask shell命令是继flask run后被实现的第二个“核心”命令，其目的是启动一个python解释器包含应用的上下文：

1
2
3
4
5
6
7
8
9
10
11
12
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end> flask shell
Python 3.10.5 (tags/v3.10.5:f377153, Jun  6 2022, 16:14:13) [MSC v.1929 64 bit (AMD64)] on win32
App: tutorweb
Instance: C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end\instance
>>> tutorweb
Traceback (most recent call last):
  File "<console>", line 1, in <module>
NameError: name 'tutorweb' is not defined
>>> app
<Flask 'tutorweb'>
>>> db
<SQLAlchemy sqlite:///C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end\tutorweb.db>

修改back-end/madblog.py，添加一个方法：

1
2
3
4
5
6
7
8
from tutorweb import create_app,db
from tutorweb.models import User

app=create_app()

@app.shell_context_processor
def make_shell_context():
    return {'db':db,'User':User}

再次运行flask shell命令：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end> flask shell
Python 3.10.5 (tags/v3.10.5:f377153, Jun  6 2022, 16:14:13) [MSC v.1929 64 bit (AMD64)] on win32
App: tutorweb
Instance: C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end\instance
>>> app
<Flask 'tutorweb'>
>>> db
<SQLAlchemy sqlite:///C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end\tutorweb.db>
>>> User
<class 'tutorweb.models.User'>
>>> u=User(username='tom',email='tom@163.com')
>>> u.set_password('123456')
>>> u.check_password('123456')
True
>>> u.check_password('654321')
False

RESTful API设计

用户资源暂时提供以下几个API：

HTTP方法 资源URL 说明
GET /api/users 返回所有用户的集合
POST /api/users 注册一个新用户
GET /api/users/<id> 返回一个用户
PUT /api/users/<id> 修改一个用户
DELETE /api/users/<id> 删除一个用户

创建tutorweb/api/users.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from tutorweb.api import bp

@bp.route('/users', methods=['POST'])
def create_user():
    '''注册一个新用户'''
    pass

@bp.route('/users', methods=['GET'])
def get_users():
    '''返回所有用户的集合'''
    pass

@bp.route('/users/<int:id>', methods=['GET'])
def get_user(id):
    '''返回一个用户'''
    pass

@bp.route('/users/<int:id>', methods=['PUT'])
def update_user(id):
    '''修改一个用户'''
    pass

@bp.route('/users/<int:id>', methods=['DELETE'])
def delete_user(id):
    '''删除一个用户'''
    pass

修改 tutorweb/api/__init__.py，在末尾添加：

1
from tutorweb.api import ping,users

用户对象转换成JSON

Flask使用的都是User实例对象，返回响应给前端时，需要传递JSON对象。

修改 tutorweb/models.py，给 User 数据模型添加 to_dict 方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from flask import url_for
...

class User(db.Model):
    ...
    def to_dict(self, include_email=False):
        data = {
            'id': self.id,
            'username': self.username,
            '_links': {
                'self': url_for('api.get_user', id=self.id)
            }
        }
        if include_email:
            data['email'] = self.email
        return data

只有当用户请求自己的数据时才包含 email，使用 include_email 标志来确定该字段是否包含在字典中。调用该方法返回字典，再用 flask.jsonify 将字典转换成 JSON 响应

用户集合转换成JSON

API 中有 POST /users 需要返回用户集合，所以还需要添加 to_collection_dict 方法。考虑到后续会创建 Post 等数据模型。

在 tutorweb/models.py 中设计一个通用类 PaginatedAPIMixin，放到User类前：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
class PaginatedAPIMixin(object):
    @staticmethod
    def to_collection_dict(query, page, per_page, endpoint, **kwargs):
        resources = query.paginate(page=page,per_page=per_page,error_out=False)
        data = {
            'items': [item.to_dict() for item in resources.items],
            '_meta': {
                'page': page,
                'per_page': per_page,
                'total_pages': resources.pages,
                'total_items': resources.total
            },
            '_links': {
                'self': url_for(endpoint, page=page, per_page=per_page,
                                **kwargs),
                'next': url_for(endpoint, page=page + 1, per_page=per_page,
                                **kwargs) if resources.has_next else None,
                'prev': url_for(endpoint, page=page - 1, per_page=per_page,
                                **kwargs) if resources.has_prev else None
            }
        }
        return data

然后，由User类继承这个类：

1
2
class User(PaginatedAPIMixin, db.Model):
    ...

JSON转换成用户对象

前端发送过来 JSON 对象，需要转换成 User 对象：

1
2
3
4
5
6
def from_dict(self, data, new_user=False):
    for field in ['username', 'email']:
        if field in data:
            setattr(self, field, data[field])
    if new_user and 'password' in data:
        self.set_password(data['password'])

错误处理

创建 tutorweb/api/errors.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from flask import jsonify
from werkzeug.http import HTTP_STATUS_CODES

def error_response(status_code, message=None):
    payload = {'error': HTTP_STATUS_CODES.get(status_code, 'Unknown error')}
    if message:
        payload['message'] = message
    response = jsonify(payload)
    response.status_code = status_code
    return response

def bad_request(message):
    #最常用的错误：400：错误的请求
    return error_response(400, message)

注册新用户

修改tutorweb/api/users.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import re 
from flask import request,jsonify,url_for
from tutorweb import db
from tutorweb.api import bp
from tutorweb.api.errors import bad_request
from tutorweb.models import User

@bp.route('/users',methods=['POST'])
def create_user():
    #注册一个新用户
    data=request.get_json()
    if not data:
        return bad_request('You must post JSON data.')

    message={}
    if 'username' not in data or not data.get('username', None):
        message['username'] = 'Please provide a valid username.'
    pattern = '^(([^<>()\[\]\\.,;:\s@"]+(\.[^<>()\[\]\\.,;:\s@"]+)*)|(".+"))@((\[[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\])|(([a-zA-Z\-0-9]+\.)+[a-zA-Z]{2,}))$'
    if 'email' not in data or not re.match(pattern, data.get('email', None)):
        message['email'] = 'Please provide a valid email address.'
    if 'password' not in data or not data.get('password', None):
        message['password'] = 'Please provide a valid password.'

    if User.query.filter_by(username=data.get('username', None)).first():
        message['username'] = 'Please use a different username.'
    if User.query.filter_by(email=data.get('email', None)).first():
        message['email'] = 'Please use a different email address.'
    if message:
        return bad_request(message)      

    user=User()
    user.from_dict(data,new_user=True)
    db.session.add(user)
    db.session.commit()
    response=jsonify(user.to_dict())
    response.status_code=201
    #HTTP协议要求201响应包含一个值为新资源URL的Location头部
    response.headers['Location']=url_for('api.get_user',id=user.id)
    return response

在C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end的目录下右键，选择在终端打开后，运行应用：

1
2
3
4
5
6
7
8
9
10
11
PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end> venv\Scripts\activate
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end> pip install --upgrade httpie
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end> flask run

 * Serving Flask app 'madblog.py'
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on http://127.0.0.1:5000
Press CTRL+C to quit
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 354-979-609

保持应用处于运行的状态后，打开另一个终端并激活环境。使用HTTPie或Postman测试API接口：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end> http POST http://localhost:5000/api/users username=lisi password=123 email=lisi@163.com

HTTP/1.1 201 CREATED
Access-Control-Allow-Origin: *
Connection: close
Content-Length: 82
Content-Type: application/json
Date: Mon, 24 Oct 2022 03:12:01 GMT
Location: /api/users/5
Server: Werkzeug/2.2.2 Python/3.10.5

{
    "_links": {
        "self": "/api/users/5"
    },
    "id": 5,
    "username": "lisi"
}

若应用没有运行，会报错：

1
2
3
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end> http POST http://localhost:5000/api/users username=lisi password=123 email=lisi@163.com

http: error: ConnectionError: HTTPConnectionPool(host='localhost', port=5000): Max retries exceeded with url: /api/users (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x0000022A6C404550>: Failed to establish a new connection: [WinError 10061] 由于目标计算机积极拒绝，无法连接。')) while doing a POST request to URL: http://localhost:5000/api/users

检索单个用户

修改tutorweb/api/users.py：

1
2
3
4
@bp.route('/users/<int:id>', methods=['GET'])
def get_user(id):
    #返回一个用户
    return jsonify(User.query.get_or_404(id).to_dict())

保持应用处于运行的状态后，在另一个终端检索用户：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end> http GET http://localhost:5000/api/users/4 

HTTP/1.1 200 OK
Access-Control-Allow-Origin: *
Connection: close
Content-Type: application/json
Date: Mon, 24 Oct 2022 03:16:47 GMT
Server: Werkzeug/2.2.2 Python/3.10.5

{
    "_links": {
        "self": "/api/users/4"
    },
    "id": 4,
    "username": "zhangsan"
}

若查询的用户id不存在，返回404错误

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end> http GET http://localhost:5000/api/users/7 

HTTP/1.1 404 NOT FOUND
Access-Control-Allow-Origin: *
Connection: close
Content-Length: 207
Content-Type: text/html; charset=utf-8
Date: Mon, 24 Oct 2022 03:19:48 GMT
Server: Werkzeug/2.2.2 Python/3.10.5

<!doctype html>
<html lang=en>
<title>404 Not Found</title>
<h1>Not Found</h1>
<p>The requested URL was not found on the server. If you entered the URL manually please check your spelling and try again.</p>

修改tutorweb/api/errors.py，设置返回JSON错误信息

1
2
3
4
5
6
7
8
9
10
11
12
from tutorweb import db
from tutorweb.api import bp
...

@bp.app_errorhandler(404)
def not_found_error(error):
    return error_response(404)

@bp.app_errorhandler(500)  
def internal_error(error):
    db.session.rollback()
    return error_response(500)

若查询的用户id不存在，此时返回404错误信息为：

1
2
3
4
5
6
7
8
9
10
11
12
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end> http GET http://localhost:5000/api/users/7
HTTP/1.1 404 NOT FOUND
Access-Control-Allow-Origin: *
Connection: close
Content-Length: 27
Content-Type: application/json
Date: Mon, 24 Oct 2022 03:17:01 GMT
Server: Werkzeug/2.2.2 Python/3.10.5

{
    "error": "Not Found"
}

检索用户集合

修改tutorweb/api/users.py：

1
2
3
4
5
6
7
@bp.route('/users',methods=['GET'])
def get_users():
    #返回所有用户的集合
    page=request.args.get('page',1,type=int)
    per_page=min(request.args.get('per_page',10,type=int),100)
    data=User.to_collection_dict(User.query,page,per_page,'api.get_users')
    return jsonify(data)

终端测试如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end> http GET http://localhost:5000/api/users 

HTTP/1.1 200 OK
Access-Control-Allow-Origin: *
Connection: close
Content-Length: 889
Content-Type: application/json
Date: Mon, 24 Oct 2022 06:24:27 GMT
Server: Werkzeug/2.2.2 Python/3.10.5

{
    "_links": {
        "next": null,
        "prev": null,
        "self": "/api/users?page=1&per_page=10"
    },
    "_meta": {
        "page": 1,
        "per_page": 10,
        "total_items": 6,
        "total_pages": 1
    },
    "items": [
        {
            "_links": {
                "self": "/api/users/1"
            },
            "id": 1,
            "username": "alice"
        },
        {
            "_links": {
                "self": "/api/users/2"
            },
            "id": 2,
            "username": "bob"
        },
        {
            "_links": {
                "self": "/api/users/3"
            },
            "id": 3,
            "username": "madman"
        },
        {
            "_links": {
                "self": "/api/users/4"
            },
            "id": 4,
            "username": "zhangsan"
        },
        {
            "_links": {
                "self": "/api/users/5"
            },
            "id": 5,
            "username": "lisi"
        },
        {
            "_links": {
                "self": "/api/users/6"
            },
            "id": 6,
            "username": "wangwu"
        }
    ]
}

修改用户

修改tutorweb/api/users.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
@bp.route('/users/<int:id>',methods=['PUT'])
def update_user(id):
    #修改一个用户
    user=User.query.get_or_404(id)
    data=request.get_json()
    if not data:
        return bad_request('You must post JSON data.')

    message={}
    if 'username' in data and not data.get('username', None):
        message['username']='Please provide a valid username.'
    pattern='^(([^<>()\[\]\\.,;:\s@"]+(\.[^<>()\[\]\\.,;:\s@"]+)*)|(".+"))@((\[[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\])|(([a-zA-Z\-0-9]+\.)+[a-zA-Z]{2,}))$'
    if 'email' in data and not re.match(pattern, data.get('email', None)):
        message['email']='Please provide a valid email address.'
    if 'username' in data and data['username'] != user.username and \
            User.query.filter_by(username=data['username']).first():
        message['username']='Please use a different username.'
    if 'email' in data and data['email'] != user.email and \
            User.query.filter_by(email=data['email']).first():
        message['email']='Please use a different email address.'

    if message:
        return bad_request(message) 
        
    user.from_dict(data,new_user=False)    
    db.session.commit()
    return jsonify(user.to_dict())

终端测试如下： 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end> http PUT http://localhost:5000/api/users/5 email=madman@gmail.com 

HTTP/1.1 200 OK
Access-Control-Allow-Origin: *
Connection: close
Content-Length: 82
Content-Type: application/json
Date: Mon, 24 Oct 2022 05:16:38 GMT
Server: Werkzeug/2.2.2 Python/3.10.5

{
    "_links": {
        "self": "/api/users/5"
    },
    "id": 5,
    "username": "lisi"
}

API认证

为了简化使用 token 认证时客户端和服务器之间的交互，可以使用 Flask-HTTPAuth 插件

1
2
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end>pip install flask-httpauth
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end>pip freeze > requirements.txt

当客户端想要开始与 API 交互时，它需要使用用户名和密码进行 Basic Auth 验证，然后获得一个临时 token。只要 token 有效，客户端就可以发送附带 token 的 API 请求以通过认证。一旦 token 到期，需要申请新的 token。

User 数据模型添加 token

修改 tutorweb/models.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import base64
from datetime import datetime, timedelta
import os
...

class User(PaginatedAPIMixin, db.Model):
    ...
    token = db.Column(db.String(32), index=True, unique=True)
    token_expiration = db.Column(db.DateTime)
    ...

    def get_token(self, expires_in=3600):
        now = datetime.utcnow()
        if self.token and self.token_expiration > now + timedelta(seconds=60):
            return self.token
        self.token = base64.b64encode(os.urandom(24)).decode('utf-8')
        self.token_expiration = now + timedelta(seconds=expires_in)
        db.session.add(self)
        return self.token

    def revoke_token(self):
        self.token_expiration = datetime.utcnow() - timedelta(seconds=1)

    @staticmethod
    def check_token(token):
        user = User.query.filter_by(token=token).first()
        if user is None or user.token_expiration < datetime.utcnow():
            return None
        return user

创建数据库迁移脚本并应用：

1
2
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end>flask db migrate -m "user add tokens"
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end>flask db upgrade

HTTP Basic Authentication

创建 tutorweb/api/auth.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from flask import g
from flask_httpauth import HTTPBasicAuth,HTTPTokenAuth
from tutorweb.models import User
from tutorweb.api.errors import error_response

basic_auth=HTTPBasicAuth()

@basic_auth.verify_password
def verify_password(username,password):
    #用于检查用户提供的用户名和密码
    user=User.query.filter_by(username=username).first()
    if user is None:
        return False
    g.current_user=user
    return user.check_password(password)

@basic_auth.error_handler
def basic_auth_error():
    #用于在认证失败情况下返回错误响应
    return error_response(401)

客户端申请 Token

目前已经实现了 Basic Auth 验证的支持，因此可以添加一条 token 检索路由，以便客户端在需要 token 时调用。

创建 tutorweb/api/tokens.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from flask import jsonify,g
from tutorweb import db
from tutorweb.api import bp
from tutorweb.api.auth import basic_auth,token_auth

@bp.route('/tokens',methods=['POST'])
@basic_auth.login_required

def get_token():
    token=g.current_user.get_token()
    db.session.commit()
    return jsonify({'token':token})

@bp.route('/tokens',methods=['DELETE'])    
@token_auth.login_required
def revoke_token():
    g.current_user.revoke_token()
    db.session.commit()
    return '',204

装饰器 @basic_auth.login_required 将指示 Flask-HTTPAuth 验证身份，当通过 Basic Auth 验证后，才使用用户模型的 get_token() 方法来生成 token，数据库提交在生成 token 后发出，以确保 token 及其到期时间被写回到数据库。

修改 tutorweb/api/__init__.py，在末尾添加：

1
from tutorweb.api import ping, users, tokens

如果尝试直接向 token API 路由发送 POST 请求，则会发生以下情况：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end> http POST http://localhost:5000/api/tokens

HTTP/1.1 401 UNAUTHORIZED
Access-Control-Allow-Origin: *
Connection: close
Content-Length: 30
Content-Type: application/json
Date: Mon, 24 Oct 2022 04:51:39 GMT
Server: Werkzeug/2.2.2 Python/3.10.5
WWW-Authenticate: Basic realm="Authentication Required"

{
    "error": "Unauthorized"
}

如果在 POST 请求附带上了 Basic Auth 需要的凭证：

1
2
3
4
5
6
7
8
9
10
11
12
13
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end> http --auth madman:123 POST http://localhost:5000/api/tokens
                
HTTP/1.1 200 OK
Access-Control-Allow-Origin: *
Connection: close
Content-Length: 50
Content-Type: application/json
Date: Mon, 24 Oct 2022 04:52:22 GMT
Server: Werkzeug/2.2.2 Python/3.10.5

{
    "token": "q2L2Umakr5/iSrf1L4mglZmYoBD/K9Je"
}

HTTP Token Authentication

用户通过 Basic Auth 拿到 token 后，之后的请求只要附带这个 token 就能够访问其它 API，修改 tutorweb/api/auth.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from flask_httpauth import HTTPBasicAuth,HTTPTokenAuth
...

token_auth=HTTPTokenAuth()
...

@token_auth.verify_token
def verify_token(token):
    #用于检查用户请求是否有token,且token真实存在，还在有效期内
    g.current_user=User.check_token(token) if token else None
    return g.current_user is not None

@token_auth.error_handler
def token_auth_error():
    #用于在Token Auth认证失败的情况下返回错误响应
    return error_response(401)

修改tutorweb/api/users.py，在文件开头导入模块：

1
from tutorweb.api.auth import token_auth

使用 Token 机制保护 API 路由

除 create_user() 之外的所有 API 视图函数需要添加 @token_auth.login_required 装饰器， create_user()函数不能使用 token 认证，因为用户不存在时不会有 token 。

1
2
3
4
5
6
7
8
9
10
11
@bp.route('/users', methods=['GET'])
@token_auth.login_required
def get_users():
    ...

@bp.route('/users/<int:id>', methods=['GET'])
@token_auth.login_required
def get_user(id):
    ...

...

若直接对上面列出的受 token 保护的 endpoint 发起请求，则会得到一个 401 错误：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end> http GET http://localhost:5000/api/users/5

HTTP/1.1 401 UNAUTHORIZED
Access-Control-Allow-Origin: *
Connection: close
Content-Length: 30
Content-Type: application/json
Date: Mon, 24 Oct 2022 04:56:51 GMT
Server: Werkzeug/2.2.2 Python/3.10.5
WWW-Authenticate: Bearer realm="Authentication Required"

{
    "error": "Unauthorized"
}

为了成功访问，需要添加 Authorization 头部，其值是请求 /api/tokens 获得的 token 的值：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end> http GET http://localhost:5000/api/users/5 "Authorization:Bearer q2L2Umakr5/iSrf1L4mglZmYoBD/K9Je"

HTTP/1.1 200 OK
Access-Control-Allow-Origin: *
Connection: close
Content-Length: 82
Content-Type: application/json
Date: Mon, 24 Oct 2022 04:58:20 GMT
Server: Werkzeug/2.2.2 Python/3.10.5

{
    "_links": {
        "self": "/api/users/5"
    },
    "id": 5,
    "username": "lisi"
}

撤销 Token

修改 tutorweb/api/tokens.py：

1
2
3
4
5
6
7
8
9
from tutorweb.api.auth import basic_auth,token_auth
...

@bp.route('/tokens',methods=['DELETE'])    
@token_auth.login_required
def revoke_token():
    g.current_user.revoke_token()
    db.session.commit()
    return '',204

客户端可以向 /api/tokens URL发送 DELETE 请求，以使 token 失效

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end> http DELETE http://localhost:5000/api/tokens "Authorization:Bearer QYnLb1pN/l4I4j2KwZWzr7+imtyekI66"

HTTP/1.1 204 NO CONTENT
Access-Control-Allow-Origin: *
Connection: close
Content-Type: text/html; charset=utf-8
Date: Mon, 24 Oct 2022 03:31:11 GMT
Server: Werkzeug/2.2.2 Python/3.10.5

(venv) PS C:\Users\Administrator\Desktop\code_learn\flask_exercise\tutorproject\back-end> http DELETE http://localhost:5000/api/tokens "Authorization:Bearer QYnLb1pN/l4I4j2KwZWzr7+imtyekI66"

HTTP/1.1 401 UNAUTHORIZED
Access-Control-Allow-Origin: *
Connection: close
Content-Type: application/json
Date: Mon, 24 Oct 2022 03:32:00 GMT
Server: Werkzeug/2.2.2 Python/3.10.5
WWW-Authenticate: Bearer realm="Authentication Required"

{
    "error": "Unauthorized"
}

至此，Flask后端已成功配置。最终的代码目录如下：

原文作者：唐抉

原文链接：http://tangmenjue.top/2022/10/24/Flask%E5%AE%9E%E6%88%98%E4%B9%8B%E6%90%AD%E5%BB%BA%E5%8D%9A%E5%AE%A2%E5%90%8E%E7%AB%AF/

发表日期：October 24th 2022, 2:36:53 pm

更新日期：October 24th 2022, 2:36:53 pm

版权声明： © 2020-2023 本文采用知识共享署名-非商业性使用 4.0 国际许可协议进行许可

CATALOG
  1. 1. 创建一个Flask RESTful API
    1. 1.1. 配置Flask
    2. 1.2. 应用工厂
    3. 1.3. 应用启动文件
    4. 1.4. 配置文件
    5. 1.5. 读取环境变量信息
    6. 1.6. 启动应用
  2. 2. Flask设计User用户相关API
    1. 2.1. 数据库
      1. 2.1.1. ORM：SQLAlchemy
      2. 2.1.2. 定义User用户数据模型
      3. 2.1.3. 第一次数据库迁移
    2. 2.2. RESTful API设计
      1. 2.2.1. 用户对象转换成JSON
      2. 2.2.2. 用户集合转换成JSON
      3. 2.2.3. JSON转换成用户对象
      4. 2.2.4. 错误处理
      5. 2.2.5. 注册新用户
      6. 2.2.6. 检索单个用户
      7. 2.2.7. 检索用户集合
      8. 2.2.8. 修改用户
    3. 2.3. API认证
      1. 2.3.1. User 数据模型添加 token
      2. 2.3.2. HTTP Basic Authentication
      3. 2.3.3. 客户端申请 Token
    4. 2.4. HTTP Token Authentication
      1. 2.4.1. 使用 Token 机制保护 API 路由
      2. 2.4.2. 撤销 Token
Total : 62
2023
2022
2020
CSS3 CentOS 7 Linux 网络管理 Docker 应用部署 数据库工具 Flask Web 后端 Web框架 Kubernetes 可视化 大屏设计 Lottie 数据库 PgSQL Vue Vue.js 前端 YAML Redis 闪退 Elasticsearch JavaScript python 企业微信 应用开发 异步IO CSS HTML 结构性思维 XAMPP 报错 设计模式
CSS3 Linux Docker 数据库工具 Web框架 大屏设计 操作系统 学习笔记 Kubernetes 可视化 数据库 Vue 前端 YAML edge JavaScript Elasticsearch Lottie 企业微信 python 前端框架 Redis 应用开发 异步IO XAMPP