RESTful API

初识 RESTful API

什么是 RESTful API

  • REST（Representational State Transfer）表现层状态转换
    • 软件架构风格定义
    • 无状态交互的客户端-服务器架构
    • 基于 HTTP 格式的请求规范
    • 使用 JSON 携带数据

RESTful API 的组成要素

  • 端点（EndPoint）- URL
  • 请求方法（Method）
    • GET：获取服务器上保存的资源
    • POST：将新的资源保存到服务器上
    • PUT：更新服务器上现有的资源
    • DELETE：删除服务器上现有的资源
  • 头部信息（Headers）- Request、Response
  • 请求数据 - 需要发送给服务器的具体数据

实现第一个 RESTful API

基于 Python3.10 实现的 flask 项目

准备一个requirements.txt

1
flask==3.0.0

新建一个/routes的 package

1
2
3
4
5
6
7
8
# -*- coding:utf-8 -*-
# Author: Zachary

from flask import Flask

app = Flask(__name__)

from routes import student_api

接着创建一个/routes/student_api.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# -*- coding:utf-8 -*-
# Author: Zachary
from flask import jsonify
from routes import app


@app.route('/student/<student_id>', method='GET')
def get_student(student_id):
    student = {
        'id': student_id,
        'name': 'Zachary',
        'gender': 'male'
    }
    return jsonify(student)

修改app.py中的内容

1
2
3
4
5
from routes import app


if __name__ == '__main__':
    app.run(debug=True)

打开 debug 模式

运行项目之后，使用 postman 调试，新建一个RESTful API集

然后新建一个GET请求，命名为Get Student by ID，在请求地址输入：http://127.0.0.1:5000/students/1

使用 flask-restful 开发 API(*)

安装 flask-restful

  • 使用 pip install flask-restful==0.3.10
  • 在requirements.txt中添加
1
2
flask==3.0.0
flask-restful==0.3.10

创建 resources

在 flask 中之前的路由管理都放在了 routes 中，在 restful api 中使用/resources进行管理

在项目路径下创建一个/resources的包

1
2
3
4
5
6
7
# -*- coding:utf-8 -*-
# Author: Zachary
from flask import Flask
from flask_restful import Api

app = Flask(__name__)
api = Api(app)

创建资源类

在包/resources下创建一个student_resource.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# -*- coding:utf-8 -*-
# Author: Zachary
from flask_restful import Resource

from resources import api


class StudentResource(Resource):
    def get(self, student_id: int):
        student = {
            'id': student_id,
            'name': 'Zachary',
            'gender': 'male'
        }
        return student


api.add_resource(StudentResource, '/students/<int:student_id>')

然后给init.py 添加一下这个 student_resource

1
2
3
4
5
6
7
8
9
# -*- coding:utf-8 -*-
# Author: Zachary
from flask import Flask
from flask_restful import Api

app = Flask(__name__)
api = Api(app)

from resources import student_resource

最后改一下 app.py 中的 app 对象

1
2
3
4
5
from resources import app


if __name__ == '__main__':
    app.run(debug=True)

增加新请求

这块暂时模拟一下实现了一个 PUT请求

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
# -*- coding:utf-8 -*-
# Author: Zachary
from flask_restful import Resource

from resources import api


class StudentResource(Resource):
    def get(self, student_id: int):
        student = {
            'id': student_id,
            'name': 'Zachary',
            'gender': 'male'
        }
        return student

    def put(self, student_id: int):
        student = {
            'id': student_id,
            'name': 'Michael',
            'gender': 'male'
        }
        return student


api.add_resource(StudentResource, '/students/<int:student_id>')

然后在 postman 新增一个请求

加入状态码

有时候的请求会有正确结果的响应，但有时候响应式错误的，因此需要一个用于告诉浏览器，该响应结果是正确的还是错误的这样一个标识符，所以使用状态码，200 默认可以不填

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
# -*- coding:utf-8 -*-
# Author: Zachary
from flask_restful import Resource

from resources import api


class StudentResource(Resource):
    def get(self, student_id: int):
        student = {
            'id': student_id,
            'name': 'Zachary',
            'gender': 'male'
        }
        if student_id == 1:
            return student, 200
        else:
            return {'error': 'Student not found'}, 404

    def put(self, student_id: int):
        student = {
            'id': student_id,
            'name': 'Michael',
            'gender': 'male'
        }
        return student


api.add_resource(StudentResource, '/students/<int:student_id>')

连接数据库

数据库准备

先准备一下数据库，创建一个名为restful_db的数据库，然后准备一个books的表

1
2
3
4
5
6
7
8
9
10
create database restful_db;

use restful_db;

create TABLE `books`(
    `id` INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
    `title` VARCHAR(255) NOT NULL UNIQUE,
    `author` VARCHAR(255) NOT NULL,
    `publish_time` TIMESTAMP NOT NULL
);

直接在 IDE 操作

然后随便准备点数据

相关包准备

1
2
3
4
5
flask==3.0.0
flask-restful==0.3.10
mysqlclient==2.1.1
SQLAlchemy==2.0.23
flask-SQLAlchemy==3.1.1

数据库连接初始化

在/resources的__init__.py中添加数据库的相关配置

1
2
3
4
5
6
7
8
9
10
11
12
13
# -*- coding:utf-8 -*-
# Author: Zachary
from flask import Flask
from flask_restful import Api
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
api = Api(app)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+mysqldb://root:980226@localhost:3306/restful_db'
db = SQLAlchemy(app)

from resources import student_resource

实体类映射

在项目目录下创建一个/models的 package

然后创建一个book.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# -*- coding:utf-8 -*-
# Author: Zachary
from datetime import datetime

from sqlalchemy import Integer, String, TIMESTAMP
from sqlalchemy.orm import mapped_column, Mapped

from resources import db


class Book(db.Model):
    __tablename__ = 'books'
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    title: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    author:Mapped[str] = mapped_column(String(255), nullable=False)
    publish_time:Mapped[datetime] = mapped_column(TIMESTAMP, nullable=False)

定义 BookService

在项目目录下创建一个/services，在其中新建一个book_service.py

1
2
3
4
5
6
7
8
9
10
# -*- coding:utf-8 -*-
# Author: Zachary
from models.book import Book
from resources import db


class BookService:

    def get_book_by_id(self, book_id: int):
        return db.session.get(Book, book_id)

定义 BookResources

在/resources目录下新建一个book_resource.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# -*- coding:utf-8 -*-
# Author: Zachary
from flask_restful import Resource

from resources import api
from services.book_service import BookService


class BookResource(Resource):
    def get(self, book_id: int):
        book = BookService().get_book_by_id(book_id)
        if book:
            return book, 200
        else:
            return {'error': f'Book(book_id:{book_id}) not found'}, 404


api.add_resource(BookResource, '/books/<int:book_id>')

然后引入一下 book_resource

1
2
3
4
5
6
7
8
9
10
11
12
# -*- coding:utf-8 -*-
# Author: Zachary
from flask import Flask
from flask_restful import Api
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
api = Api(app)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+mysqldb://root:980226@localhost:3306/restful_db'
db = SQLAlchemy(app)

from resources import student_resource, book_resource

这时候如果尝试去测试一下会报错

1
2
    raise TypeError(f'Object of type {o.__class__.__name__} '
TypeError: Object of type Book is not JSON serializable

解决方法：使用一个序列化方法，返回对象，定义了一个serialize方法，其中 publish_time 也需要再进一步序列化，使用isoformat()将其转换为字符串

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
# -*- coding:utf-8 -*-
# Author: Zachary
from datetime import datetime

from sqlalchemy import Integer, String, TIMESTAMP
from sqlalchemy.orm import mapped_column, Mapped

from resources import db


class Book(db.Model):
    __tablename__ = 'books'
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    title: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    author: Mapped[str] = mapped_column(String(255), nullable=False)
    publish_time: Mapped[datetime] = mapped_column(TIMESTAMP, nullable=False)

    def serialize(self):
        return {
            'id': self.id,
            'title': self.title,
            'author': self.author,
            'publish_time': self.publish_time.isoformat()
        }

然后book_resource也稍微修改

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# -*- coding:utf-8 -*-
# Author: Zachary
from flask_restful import Resource

from resources import api
from services.book_service import BookService


class BookResource(Resource):
    def get(self, book_id: int):
        book = BookService().get_book_by_id(book_id)
        if book:
            return book.serialize(), 200
        else:
            return {'error': f'Book(book_id:{book_id}) not found'}, 404


api.add_resource(BookResource, '/books/<int:book_id>')

结果：

新增与修改 API

新增数据 API

定义资源类

新增数据需要使用POST来实现，考虑到新增的时候 URL 路径不需要传递参数

但是为了区分先前定义的资源，同时可以把查找所有书籍也实现一下

重新定义一个资源类BookListResource

  • 获取所有书籍列表
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
# -*- coding:utf-8 -*-
# Author: Zachary
from flask_restful import Resource

from resources import api
from services.book_service import BookService


class BookResource(Resource):
    def get(self, book_id: int):
        book = BookService().get_book_by_id(book_id)
        if book:
            return book.serialize(), 200
        else:
            return {'error': f'Book(book_id:{book_id}) not found'}, 404


class BookListResource(Resource):
    def get(self):
        book_list = BookService().get_all_books()
        return [book.serialize() for book in book_list]


api.add_resource(BookResource, '/books/<int:book_id>')
api.add_resource(BookListResource, '/books')

同时得实现相关的 Service，包括查询所有书籍

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# -*- coding:utf-8 -*-
# Author: Zachary
from sqlalchemy import asc, Select

from models.book import Book
from resources import db


class BookService:

    def get_book_by_id(self, book_id: int):
        return db.session.get(Book, book_id)

    def get_all_books(self):
        query = Select(Book).order_by(asc(Book.title))
        return db.session.scalars(query).all()

测试一下获取所有书籍

  • 新增一个书籍
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
# -*- coding:utf-8 -*-
# Author: Zachary
from datetime import datetime

from flask import request
from flask_restful import Resource

from models.book import Book
from resources import api
from services.book_service import BookService


class BookResource(Resource):
    def get(self, book_id: int):
        book = BookService().get_book_by_id(book_id)
        if book:
            return book.serialize(), 200
        else:
            return {'error': f'Book(book_id:{book_id}) not found'}, 404


class BookListResource(Resource):
    def get(self):
        book_list = BookService().get_all_books()
        return [book.serialize() for book in book_list]

    def post(self):
        request_json = request.json
        if request_json:
            title = request_json.get('title', None)
            author = request_json.get('author', None)
            publish_time = datetime.fromisoformat(request_json.get('publish_time', None))
            book = Book(title=title, author=author, publish_time=publish_time)

            if title and author and publish_time:
                book = BookService().create_book(book)
                return book.serialize(), 200
        else:
            return {'error': 'Invalid request, please provide book info as json'}, 400


api.add_resource(BookResource, '/books/<int:book_id>')
api.add_resource(BookListResource, '/books')
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# -*- coding:utf-8 -*-
# Author: Zachary
from sqlalchemy import asc, Select

from models.book import Book
from resources import db


class BookService:

    def get_book_by_id(self, book_id: int):
        return db.session.get(Book, book_id)

    def get_all_books(self):
        query = Select(Book).order_by(asc(Book.title))
        return db.session.scalars(query).all()

    def create_book(self, book: Book):
        db.session.add(book)
        db.session.commit()
        return book

测试一下 新增一本书籍

再使用获取书籍 获取全部书籍

修复重复提交错误显示

但是如果现在重复提交的话，由于书名不允许重复就会出现这样的错误

应该修改地 显示更加直观

其实就是在新增的时候先查一下 该 title 是不是在数据库中存在 然后做个判断

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
# -*- coding:utf-8 -*-
# Author: Zachary
from datetime import datetime

from flask import request
from flask_restful import Resource

from models.book import Book
from resources import api
from services.book_service import BookService


class BookResource(Resource):
    def get(self, book_id: int):
        book = BookService().get_book_by_id(book_id)
        if book:
            return book.serialize(), 200
        else:
            return {'error': f'Book(book_id:{book_id}) not found'}, 404


class BookListResource(Resource):
    def get(self):
        book_list = BookService().get_all_books()
        return [book.serialize() for book in book_list]

    def post(self):
        try:
            request_json = request.json
            if request_json:
                title = request_json.get('title', None)
                author = request_json.get('author', None)
                publish_time = datetime.fromisoformat(request_json.get('publish_time', None))
                book = Book(title=title, author=author, publish_time=publish_time)

                if title and author and publish_time:
                    book = BookService().create_book(book)
                    return book.serialize(), 200
            else:
                return {'error': 'Invalid request, please provide book info as json'}, 400
        except Exception as e:
            return {'error': f'{e}'}, 500


api.add_resource(BookResource, '/books/<int:book_id>')
api.add_resource(BookListResource, '/books')
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
# -*- coding:utf-8 -*-
# Author: Zachary
from sqlalchemy import asc, Select

from models.book import Book
from resources import db


class BookService:

    def get_book_by_id(self, book_id: int):
        return db.session.get(Book, book_id)

    def get_all_books(self):
        query = Select(Book).order_by(asc(Book.title))
        return db.session.scalars(query).all()

    def get_book_by_title(self, title: str):
        query = Select(Book).where(Book.title == title)
        return db.session.scalar(query)

    def create_book(self, book: Book):
        existed_book = self.get_book_by_title(book.title)
        if existed_book:
            raise Exception(f'Book(title:{book.title}) already exists')
        db.session.add(book)
        db.session.commit()
        return book

这时 如果再重复提交会

修改数据 API

修改数据这块的逻辑其实和上面新增数据差不多，代码大多可以复用

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
# -*- coding:utf-8 -*-
# Author: Zachary
from datetime import datetime

from flask import request
from flask_restful import Resource

from models.book import Book
from resources import api
from services.book_service import BookService


class BookResource(Resource):
    def get(self, book_id: int):
        book = BookService().get_book_by_id(book_id)
        if book:
            return book.serialize(), 200
        else:
            return {'error': f'Book(book_id:{book_id}) not found'}, 404

    def put(self, book_id: int):
        try:
            request_json = request.json
            if request_json:
                title = request_json.get('title', None)
                author = request_json.get('author', None)
                publish_time_str = request_json.get('publish_time', None)
                publish_time = datetime.fromisoformat(publish_time_str) if publish_time_str else None
                book = Book(id=book_id, title=title, author=author, publish_time=publish_time)

                book = BookService().update_book(book)
                return book.serialize(), 200
            else:
                return {'error': 'Invalid request, please provide book info as json'}, 400
        except Exception as e:
            return {'error': f'{e}'}, 500


class BookListResource(Resource):
    def get(self):
        book_list = BookService().get_all_books()
        return [book.serialize() for book in book_list]

    def post(self):
        try:
            request_json = request.json
            if request_json:
                title = request_json.get('title', None)
                author = request_json.get('author', None)
                publish_time_str = request_json.get('publish_time', None)
                publish_time = datetime.fromisoformat(publish_time_str) if publish_time_str else None

                if title and author and publish_time:
                    book = Book(title=title, author=author, publish_time=publish_time)
                    book = BookService().create_book(book)
                    return book.serialize(), 200
                else:
                    return {'error': 'Invalid request, please provide complete book info'}, 400
            else:
                return {'error': 'Invalid request, please provide book info as json'}, 400
        except Exception as e:
            return {'error': f'{e}'}, 500


api.add_resource(BookResource, '/books/<int:book_id>')
api.add_resource(BookListResource, '/books')
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
# -*- coding:utf-8 -*-
# Author: Zachary
from sqlalchemy import asc, Select, and_

from models.book import Book
from resources import db


class BookService:

    def get_book_by_id(self, book_id: int):
        return db.session.get(Book, book_id)

    def get_all_books(self):
        query = Select(Book).order_by(asc(Book.title))
        return db.session.scalars(query).all()

    def get_book_by_title(self, title: str):
        query = Select(Book).where(Book.title == title)
        return db.session.scalar(query)

    def create_book(self, book: Book):
        existed_book = self.get_book_by_title(book.title)
        if existed_book:
            raise Exception(f'Book(title:{book.title}) already exists')
        db.session.add(book)
        db.session.commit()
        return book

    def get_book_by_title_and_not_id(self, title: str, book_id: int):
        query = Select(Book).where(and_(Book.title == title, Book.id != book_id))
        return db.session.scalar(query)

    def update_book(self, book: Book):
        existed_book = self.get_book_by_id(book.id)
        if not existed_book:
            raise Exception(f'Book(id:{book.id}) not found')
        existed_same_title_book = self.get_book_by_title_and_not_id(book.title, book.id)
        if existed_same_title_book:
            raise Exception(f'Book(title:{book.title}) already exists')
        if book.title:
            existed_book.title = book.title
        if book.author:
            existed_book.author = book.author
        if book.publish_time:
            existed_book.publish_time = book.publish_time
        db.session.commit()
        return existed_book

API 身份认证

为什么需要做身份认证

通常情况，开发出来的 API 肯定是给特定客户才能使用的，比如注册过的用户、会员等，只有符合身份或者权限的客户才能调用 API，并不希望谁都可以使用

如何实现的身份认证：

画板

包准备工作

需要使用到PyJWT

1
2
3
4
5
6
flask==3.0.0
flask-restful==0.3.10
mysqlclient==2.1.1
SQLAlchemy==2.0.23
flask-SQLAlchemy==3.1.1
PyJWT==2.8.0

准备用户表

首先创建一张users表

1
2
3
4
5
create TABLE `users`(
    `id` INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
    `username` VARCHAR(128) NOT NULL UNIQUE,
    `password` VARCHAR(128) NOT NULL
);

然后随便准备一条数据

实体类映射

在/models下创建一个user.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# -*- coding:utf-8 -*-
# Author: Zachary
from sqlalchemy import Integer, String
from sqlalchemy.orm import Mapped, mapped_column

from resources import db


class User(db.Model):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    username: Mapped[str] = mapped_column(String(128), unique=True, nullable=False)
    password: Mapped[str] = mapped_column(String(128), nullable=False)

    def serialize(self):
        return {
            'id': self.id,
            'username': self.username
        }

定义 UserService

在/services下创建一个user_service.py

1
2
3
4
5
6
7
8
9
10
11
12
# -*- coding:utf-8 -*-
# Author: Zachary
from sqlalchemy import Select, and_

from models.user import User
from resources import db


class UserService:
    def login(self, username: str, password: str):
        query = Select(User).where(and_(User.username == username, User.password == password))
        return db.session.scalar(query)

定义 UserResource

在/resources下创建一个user_resource.py

  • 登录一般多用的是POST请求，虽然感觉是像是去查数据库应该用 GET
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
# -*- coding:utf-8 -*-
# Author: Zachary
from flask import request
from flask_restful import Resource

from resources import api
from services.user_service import UserService


class LoginResource(Resource):
    def post(self):
        try:
            request_json = request.json
            if request_json:
                username = request_json.get('username', None)
                password = request_json.get('password', None)

                if username and password:
                    user = UserService().login(username, password)
                    if user:
                        return user.serialize(), 200
                    else:
                        return {'error': 'Invalid username or password'}, 401
                else:
                    return {'error': 'Invalid request, please provide complete username and password'}, 400
            else:
                return {'error': 'Invalid request, please provide username and password info as json'}, 400
        except Exception as e:
            return {'error': f'{e}'}, 500


api.add_resource(LoginResource, '/login')

同时还有 user_resource 需要添加一下

1
2
3
4
5
6
7
8
9
10
11
12
# -*- coding:utf-8 -*-
# Author: Zachary
from flask import Flask
from flask_restful import Api
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
api = Api(app)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+mysqldb://root:980226@localhost:3306/restful_db'
db = SQLAlchemy(app)

from resources import student_resource, book_resource, user_resource

写到这里可以尝试测试一下登录的功能

返回加密 token

登录秘钥

先在项目目录下创建一个/common的 package，然后在其中添加一个constants.py，用于保存常量

创建一个LOGIN_SECRET

1
2
3
4
# -*- coding:utf-8 -*-
# Author: Zachary

LOGIN_SECRET = 'zacharyloginsecret'

加入 token

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
# -*- coding:utf-8 -*-
# Author: Zachary
import jwt
from flask import request
from flask_restful import Resource

from common.constants import LOGIN_SECRET
from resources import api
from services.user_service import UserService


class LoginResource(Resource):
    def post(self):
        try:
            request_json = request.json
            if request_json:
                username = request_json.get('username', None)
                password = request_json.get('password', None)

                if username and password:
                    user = UserService().login(username, password)
                    if user:
                        user_json = user.serialize()
                        jwt_token = jwt.encode(user_json, LOGIN_SECRET, algorithm='HS256')
                        user_json['token'] = jwt_token
                        return user_json, 200
                    else:
                        return {'error': 'Invalid username or password'}, 401
                else:
                    return {'error': 'Invalid request, please provide complete username and password'}, 400
            else:
                return {'error': 'Invalid request, please provide username and password info as json'}, 400
        except Exception as e:
            return {'error': f'{e}'}, 500


api.add_resource(LoginResource, '/login')

添加书籍需授权才可操作

现在应用一下上面生成的这个 token

现在用户需要登录了才可以使用 添加书籍 功能

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
68
69
70
71
72
73
74
75
76
77
78
79
# -*- coding:utf-8 -*-
# Author: Zachary
from datetime import datetime

import jwt
from flask import request
from flask_restful import Resource

from common.constants import LOGIN_SECRET
from models.book import Book
from resources import api
from services.book_service import BookService


class BookResource(Resource):
    def get(self, book_id: int):
        book = BookService().get_book_by_id(book_id)
        if book:
            return book.serialize(), 200
        else:
            return {'error': f'Book(book_id:{book_id}) not found'}, 404

    def put(self, book_id: int):
        try:
            request_json = request.json
            if request_json:
                title = request_json.get('title', None)
                author = request_json.get('author', None)
                publish_time_str = request_json.get('publish_time', None)
                publish_time = datetime.fromisoformat(publish_time_str) if publish_time_str else None
                book = Book(id=book_id, title=title, author=author, publish_time=publish_time)

                book = BookService().update_book(book)
                return book.serialize(), 200
            else:
                return {'error': 'Invalid request, please provide book info as json'}, 400
        except Exception as e:
            return {'error': f'{e}'}, 500


class BookListResource(Resource):
    def get(self):
        book_list = BookService().get_all_books()
        return [book.serialize() for book in book_list]

    def post(self):
        # 添加鉴权的部分
        jwt_token = request.headers.get('token', None)
        if not jwt_token:
            return {'error': 'Unauthorized, please provide token'}, 401
        try:
            user_info = jwt.decode(jwt_token, LOGIN_SECRET, algorithms=['HS256'])
            if not user_info or not user_info.get('username', None):
                raise Exception('Unauthorized, invalid token')
        except Exception as e:
            return {'error': 'Unauthorized, invalid token'}, 401

        try:
            request_json = request.json
            if request_json:
                title = request_json.get('title', None)
                author = request_json.get('author', None)
                publish_time_str = request_json.get('publish_time', None)
                publish_time = datetime.fromisoformat(publish_time_str) if publish_time_str else None

                if title and author and publish_time:
                    book = Book(title=title, author=author, publish_time=publish_time)
                    book = BookService().create_book(book)
                    return book.serialize(), 200
                else:
                    return {'error': 'Invalid request, please provide complete book info'}, 400
            else:
                return {'error': 'Invalid request, please provide book info as json'}, 400
        except Exception as e:
            return {'error': f'{e}'}, 500


api.add_resource(BookResource, '/books/<int:book_id>')
api.add_resource(BookListResource, '/books')

测试一下

需要在 request 的 headers 中添加上 token

如果不提供

装饰器复用加密授权

上面的代码可以看出 如果要加密授权需要在每一个 resource 代码的前面写上校验的代码，如果一直复制粘贴也很复杂；

使用装饰器进行一个代码复用

在/common的包下创建一个api_tools.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
# -*- coding:utf-8 -*-
# Author: Zachary
from functools import wraps

import jwt
from flask import request

from common.constants import LOGIN_SECRET


def token_required():
    def check_token(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            token = request.headers.get('token', None)
            if not token:
                return {'error': 'Unauthorized, please provide token'}, 401
            try:
                user_info = jwt.decode(token, LOGIN_SECRET, algorithms=['HS256'])
                if not user_info or not user_info.get('username', None):
                    raise Exception('Unauthorized, invalid token')
            except Exception as e:
                return {'error': 'Unauthorized, invalid token'}, 401
            result = f(*args, **kwargs)
            return result

        return wrapper

    return check_token

需要使用身份验证的代码 打上这个装饰器就可以了，并且尝试着给修改书籍信息也加上这个身份验证

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
68
69
70
71
72
# -*- coding:utf-8 -*-
# Author: Zachary
from datetime import datetime

import jwt
from flask import request
from flask_restful import Resource

from common.api_tools import token_required
from common.constants import LOGIN_SECRET
from models.book import Book
from resources import api
from services.book_service import BookService


class BookResource(Resource):
    def get(self, book_id: int):
        book = BookService().get_book_by_id(book_id)
        if book:
            return book.serialize(), 200
        else:
            return {'error': f'Book(book_id:{book_id}) not found'}, 404

    @token_required()
    def put(self, book_id: int):
        try:
            request_json = request.json
            if request_json:
                title = request_json.get('title', None)
                author = request_json.get('author', None)
                publish_time_str = request_json.get('publish_time', None)
                publish_time = datetime.fromisoformat(publish_time_str) if publish_time_str else None
                book = Book(id=book_id, title=title, author=author, publish_time=publish_time)

                book = BookService().update_book(book)
                return book.serialize(), 200
            else:
                return {'error': 'Invalid request, please provide book info as json'}, 400
        except Exception as e:
            return {'error': f'{e}'}, 500


class BookListResource(Resource):
    def get(self):
        book_list = BookService().get_all_books()
        return [book.serialize() for book in book_list]

    @token_required()
    def post(self):
        try:
            request_json = request.json
            if request_json:
                title = request_json.get('title', None)
                author = request_json.get('author', None)
                publish_time_str = request_json.get('publish_time', None)
                publish_time = datetime.fromisoformat(publish_time_str) if publish_time_str else None

                if title and author and publish_time:
                    book = Book(title=title, author=author, publish_time=publish_time)
                    book = BookService().create_book(book)
                    return book.serialize(), 200
                else:
                    return {'error': 'Invalid request, please provide complete book info'}, 400
            else:
                return {'error': 'Invalid request, please provide book info as json'}, 400
        except Exception as e:
            return {'error': f'{e}'}, 500


api.add_resource(BookResource, '/books/<int:book_id>')
api.add_resource(BookListResource, '/books')

测试修改书籍信息

操作文件 API

文件上传

RequestParser 处理请求参数输入

  • parser = reqparse.RequestParser
    • parser.add_argument("请求参数", required=True, type=FileStorage, location="files", help="please provide file")
      • "请求参数" - 放置的是 resource 的请求参数
      • required - 请求参数是否是必须得
      • type - 要将请求参数转换成的的数据类型
      • location - 请求分析器从哪里提取参数
        • files - 文件
        • headers - 请求头
        • args - 请求参数
      • help - 如果请求不包含"请求参数"会提示的报错信息

定义 Resource

首先在/resources下创建一个attachment_resource.py

然后需要在__init__()里面使用reqparse模块的语法分析器进行请求分析，自动处理请求中的参数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# -*- coding:utf-8 -*-
# Author: Zachary
from flask_restful import Resource, reqparse
from werkzeug.datastructures import FileStorage

from common.utils import get_attachment_path
from resources import api


class AttachmentListResource(Resource):
    def __init__(self):
        self.parser = reqparse.RequestParser()
        self.parser.add_argument('attachment', required=True, type=FileStorage, location='files',
                                 help='please provide attachment file')

    def post(self):
        attachment_file = self.parser.parse_args().get('attachment')
        file_path = get_attachment_path()
        file_save_path = file_path.joinpath(attachment_file.filename)
        attachment_file.save(file_save_path)
        return {'message': 'attachment uploaded'}


api.add_resource(AttachmentListResource, '/attachments')

设置上传的文件存储的位置，/common下创建utils.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# -*- coding:utf-8 -*-
# Author: Zachary
from pathlib import Path


def get_attachment_path():
    """
    获得项目目录下的attachements文件夹的路径
    该文件夹用于存储所上传的附件
    :return:
    """
    project_path = Path(__file__).parent.parent
    attachment_path = project_path.joinpath('attachments')

    if not attachment_path.exists():
        attachment_path.mkdir(parents=True)

    return attachment_path

并且要将attachment_resource加入配置

1
2
3
4
5
6
7
8
9
10
11
12
13
# -*- coding:utf-8 -*-
# Author: Zachary
from flask import Flask
from flask_restful import Api
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
api = Api(app)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+mysqldb://root:980226@localhost:3306/restful_db'
db = SQLAlchemy(app)

from resources import student_resource, book_resource, user_resource, attachment_resource

测试一下 上传文件

需要在 body - from-data - 设置 key-value 记住选择 File

可以看到成功上传了

文件下载

仅需使用send_file()即可

定义 Resource

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
# -*- coding:utf-8 -*-
# Author: Zachary
from flask import send_file
from flask_restful import Resource, reqparse
from werkzeug.datastructures import FileStorage

from common.utils import get_attachment_path
from resources import api


class AttachmentListResource(Resource):
    def __init__(self):
        self.parser = reqparse.RequestParser()
        self.parser.add_argument('attachment', required=True, type=FileStorage, location='files',
                                 help='please provide attachment file')

    def post(self):
        attachment_file = self.parser.parse_args().get('attachment')
        file_path = get_attachment_path()
        file_save_path = file_path.joinpath(attachment_file.filename)
        attachment_file.save(file_save_path)
        return {'message': 'attachment uploaded'}


class AttachmentResource(Resource):
    def get(self, filename):
        file_path = get_attachment_path()
        file_save_path = file_path.joinpath(filename)
        return send_file(file_save_path)


api.add_resource(AttachmentListResource, '/attachments')
api.add_resource(AttachmentResource, '/attachments/<str:filename>')

这块新上传一张图片测试一下

上传和下载都测试成功

Swagger 文档

生成文档

所需插件

  • apispec
  • flask-_apispec
  • PyYAML
1
2
3
4
5
6
7
8
9
10
flask==3.0.0
flask-restful==0.3.10
mysqlclient==2.1.1
SQLAlchemy==2.0.23
flask-SQLAlchemy==3.1.1
PyJWT==2.8.0
apispec==6.3.1
flask_apispec==0.11.4
PyYAML==6.0.1
flask_restful_swagger==0.20.2

初始化组件

在构建 api 对象的时候，进行一个配置

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
# -*- coding:utf-8 -*-
# Author: Zachary
from apispec import APISpec
from apispec.ext.marshmallow import MarshmallowPlugin
from flask import Flask
from flask_apispec import FlaskApiSpec
from flask_restful import Api
from flask_restful_swagger import swagger
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
api = swagger.docs(Api(app), apiVersion='0.1')  # swagger docs
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+mysqldb://root:980226@localhost:3306/restful_db'
app.config.update(
    {
        'APISPEC_SPEC': APISpec(
            title='RESTful API Swagger Docs',
            version='v1',
            plugins=[MarshmallowPlugin()],
            openapi_version='2.0.0'
        ),
        'APISPEC_SWAGGER_URL': '/swagger/',
        'APISPEC_SWAGGER_UI_URL': '/swagger-ui/'
    }
)
db = SQLAlchemy(app)
docs = FlaskApiSpec(app)

from resources import student_resource, book_resource, user_resource, attachment_resource

给 Resource 增加 doc 注解

要给相应的 resource 下的 api 增加 doc 文档，

  • 首先 resource 类需要 继承一个MethodResource
  • 给需要生成 doc 的 api 方法 增加一个@doc()注解
    • description 该 api 的一个描述
    • tags 分组
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
68
69
70
71
72
73
74
75
# -*- coding:utf-8 -*-
# Author: Zachary
from datetime import datetime

import jwt
from flask import request
from flask_apispec import MethodResource, doc
from flask_restful import Resource

from common.api_tools import token_required
from common.constants import LOGIN_SECRET
from models.book import Book
from resources import api, docs
from services.book_service import BookService


class BookResource(MethodResource, Resource):
    @doc(description='Get book by id', tags=['Book Requests'])
    def get(self, book_id: int):
        book = BookService().get_book_by_id(book_id)
        if book:
            return book.serialize(), 200
        else:
            return {'error': f'Book(book_id:{book_id}) not found'}, 404

    @token_required()
    def put(self, book_id: int):
        try:
            request_json = request.json
            if request_json:
                title = request_json.get('title', None)
                author = request_json.get('author', None)
                publish_time_str = request_json.get('publish_time', None)
                publish_time = datetime.fromisoformat(publish_time_str) if publish_time_str else None
                book = Book(id=book_id, title=title, author=author, publish_time=publish_time)

                book = BookService().update_book(book)
                return book.serialize(), 200
            else:
                return {'error': 'Invalid request, please provide book info as json'}, 400
        except Exception as e:
            return {'error': f'{e}'}, 500


class BookListResource(Resource):
    def get(self):
        book_list = BookService().get_all_books()
        return [book.serialize() for book in book_list]

    @token_required()
    def post(self):
        try:
            request_json = request.json
            if request_json:
                title = request_json.get('title', None)
                author = request_json.get('author', None)
                publish_time_str = request_json.get('publish_time', None)
                publish_time = datetime.fromisoformat(publish_time_str) if publish_time_str else None

                if title and author and publish_time:
                    book = Book(title=title, author=author, publish_time=publish_time)
                    book = BookService().create_book(book)
                    return book.serialize(), 200
                else:
                    return {'error': 'Invalid request, please provide complete book info'}, 400
            else:
                return {'error': 'Invalid request, please provide book info as json'}, 400
        except Exception as e:
            return {'error': f'{e}'}, 500


api.add_resource(BookResource, '/books/<int:book_id>')
api.add_resource(BookListResource, '/books')

docs.register(BookResource)

在浏览器输入

  • 127.0.0.1:5000/swagger/
  • 127.0.0.1:5000/swagger-ui/

tags 分组

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
68
69
70
71
72
73
74
75
76
77
# -*- coding:utf-8 -*-
# Author: Zachary
from datetime import datetime

import jwt
from flask import request
from flask_apispec import MethodResource, doc
from flask_restful import Resource

from common.api_tools import token_required
from common.constants import LOGIN_SECRET
from models.book import Book
from resources import api, docs
from services.book_service import BookService


class BookResource(MethodResource, Resource):
    @doc(description='Get a book by book id', tags=['Book Requests'])
    def get(self, book_id: int):
        book = BookService().get_book_by_id(book_id)
        if book:
            return book.serialize(), 200
        else:
            return {'error': f'Book(book_id:{book_id}) not found'}, 404

    @doc(description='Update a book by book id', tags=['Book Requests'])
    @token_required()
    def put(self, book_id: int):
        try:
            request_json = request.json
            if request_json:
                title = request_json.get('title', None)
                author = request_json.get('author', None)
                publish_time_str = request_json.get('publish_time', None)
                publish_time = datetime.fromisoformat(publish_time_str) if publish_time_str else None
                book = Book(id=book_id, title=title, author=author, publish_time=publish_time)

                book = BookService().update_book(book)
                return book.serialize(), 200
            else:
                return {'error': 'Invalid request, please provide book info as json'}, 400
        except Exception as e:
            return {'error': f'{e}'}, 500


class BookListResource(Resource):
    def get(self):
        book_list = BookService().get_all_books()
        return [book.serialize() for book in book_list]

    @token_required()
    def post(self):
        try:
            request_json = request.json
            if request_json:
                title = request_json.get('title', None)
                author = request_json.get('author', None)
                publish_time_str = request_json.get('publish_time', None)
                publish_time = datetime.fromisoformat(publish_time_str) if publish_time_str else None

                if title and author and publish_time:
                    book = Book(title=title, author=author, publish_time=publish_time)
                    book = BookService().create_book(book)
                    return book.serialize(), 200
                else:
                    return {'error': 'Invalid request, please provide complete book info'}, 400
            else:
                return {'error': 'Invalid request, please provide book info as json'}, 400
        except Exception as e:
            return {'error': f'{e}'}, 500


api.add_resource(BookResource, '/books/<int:book_id>')
api.add_resource(BookListResource, '/books')

docs.register(BookResource)

使用 YAML 格式

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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
# -*- coding:utf-8 -*-
# Author: Zachary
from datetime import datetime

import jwt
from flask import request, Response
from flask_apispec import MethodResource, doc
from flask_restful import Resource

from common.api_tools import token_required
from common.constants import LOGIN_SECRET
from models.book import Book
from resources import api, docs, app
from services.book_service import BookService


class BookResource(MethodResource, Resource):
    @doc(description='Get a book by book id', tags=['Book Requests'])
    def get(self, book_id: int):
        book = BookService().get_book_by_id(book_id)
        if book:
            return book.serialize(), 200
        else:
            return {'error': f'Book(book_id:{book_id}) not found'}, 404

    @doc(description='Update a book by book id', tags=['Book Requests'])
    @token_required()
    def put(self, book_id: int):
        try:
            request_json = request.json
            if request_json:
                title = request_json.get('title', None)
                author = request_json.get('author', None)
                publish_time_str = request_json.get('publish_time', None)
                publish_time = datetime.fromisoformat(publish_time_str) if publish_time_str else None
                book = Book(id=book_id, title=title, author=author, publish_time=publish_time)

                book = BookService().update_book(book)
                return book.serialize(), 200
            else:
                return {'error': 'Invalid request, please provide book info as json'}, 400
        except Exception as e:
            return {'error': f'{e}'}, 500


class BookListResource(Resource):
    def get(self):
        book_list = BookService().get_all_books()
        return [book.serialize() for book in book_list]

    @token_required()
    def post(self):
        try:
            request_json = request.json
            if request_json:
                title = request_json.get('title', None)
                author = request_json.get('author', None)
                publish_time_str = request_json.get('publish_time', None)
                publish_time = datetime.fromisoformat(publish_time_str) if publish_time_str else None

                if title and author and publish_time:
                    book = Book(title=title, author=author, publish_time=publish_time)
                    book = BookService().create_book(book)
                    return book.serialize(), 200
                else:
                    return {'error': 'Invalid request, please provide complete book info'}, 400
            else:
                return {'error': 'Invalid request, please provide book info as json'}, 400
        except Exception as e:
            return {'error': f'{e}'}, 500


api.add_resource(BookResource, '/books/<int:book_id>')
api.add_resource(BookListResource, '/books')

docs.register(BookResource)


@app.route('/swagger.yaml')
def generate_swagger_yaml():
    yaml_doc = docs.spec.to_yaml()
    return Response(yaml_doc, mimetype='text/yaml')

在浏览器访问 127.0.0.1:5000/swagger.yaml

Swagger 中的请求与响应

定义 request 中的参数（schema 类）

对之前所做的 book_resource 中的修改书籍信息的 put 方法为例

body 参数

  • 在book_resource.py中定义一个类BookRequestSchema，用于指定请求所需要的参数
1
2
3
4
class BookRequestSchema(Schema):
    title = fields.String(required=True)
    author = fields.String(requird=True)
    publish_time = fields.DateTime(required=True)
  • 然后还需要使用@use_kwargs()注解，来代替之前通过 request.json 获取的请求参数
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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
# -*- coding:utf-8 -*-
# Author: Zachary
from datetime import datetime

import jwt
from flask import request, Response
from flask_apispec import MethodResource, doc, use_kwargs
from flask_restful import Resource
from marshmallow import Schema, fields

from common.api_tools import token_required
from common.constants import LOGIN_SECRET
from models.book import Book
from resources import api, docs, app
from services.book_service import BookService


class BookRequestSchema(Schema):
    title = fields.String(required=True)
    author = fields.String(requird=True)
    publish_time = fields.DateTime(required=True)


class BookResource(MethodResource, Resource):
    @doc(description='Get a book by book id', tags=['Book Requests'])
    def get(self, book_id: int):
        book = BookService().get_book_by_id(book_id)
        if book:
            return book.serialize(), 200
        else:
            return {'error': f'Book(book_id:{book_id}) not found'}, 404

    @doc(description='Update a book by book id', tags=['Book Requests'])
    @use_kwargs(BookRequestSchema, location='json')
    @token_required()
    def put(self, book_id: int, **kwargs):
        try:
            title = kwargs.get('title', None)
            author = kwargs.get('author', None)
            publish_time = kwargs.get('publish_time', None)
            book = Book(id=book_id, title=title, author=author, publish_time=publish_time)

            book = BookService().update_book(book)
            return book.serialize(), 200
        except Exception as e:
            return {'error': f'{e}'}, 500


class BookListResource(Resource):
    def get(self):
        book_list = BookService().get_all_books()
        return [book.serialize() for book in book_list]

    @token_required()
    def post(self):
        try:
            request_json = request.json
            if request_json:
                title = request_json.get('title', None)
                author = request_json.get('author', None)
                publish_time_str = request_json.get('publish_time', None)
                publish_time = datetime.fromisoformat(publish_time_str) if publish_time_str else None

                if title and author and publish_time:
                    book = Book(title=title, author=author, publish_time=publish_time)
                    book = BookService().create_book(book)
                    return book.serialize(), 200
                else:
                    return {'error': 'Invalid request, please provide complete book info'}, 400
            else:
                return {'error': 'Invalid request, please provide book info as json'}, 400
        except Exception as e:
            return {'error': f'{e}'}, 500


api.add_resource(BookResource, '/books/<int:book_id>')
api.add_resource(BookListResource, '/books')

docs.register(BookResource)


@app.route('/swagger.yaml')
def generate_swagger_yaml():
    yaml_doc = docs.spec.to_yaml()
    return Response(yaml_doc, mimetype='text/yaml')

可以直观地看到请求体所需要的参数是什么

但是由于这个PUT是需要身份验证的，如果这时候测试一下会报错

headers 参数

所以还需要单独给身份验证，即 headers 中的参数做一个类

  • 对于 headers 中的 token 也需要单独定义一个TokenSchema类
1
2
class TokenSchema(Schema):
    token = fields.String(required=True)
  • 同样为了在 resource 中的 api 中实现对应参数，可以再一次使用@use_kwargs()注解
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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
# -*- coding:utf-8 -*-
# Author: Zachary
from datetime import datetime

import jwt
from flask import request, Response
from flask_apispec import MethodResource, doc, use_kwargs
from flask_restful import Resource
from marshmallow import Schema, fields

from common.api_tools import token_required
from common.constants import LOGIN_SECRET
from models.book import Book
from resources import api, docs, app
from services.book_service import BookService


class TokenSchema(Schema):
    token = fields.String(required=True)


class BookRequestSchema(Schema):
    title = fields.String(required=True)
    author = fields.String(requird=True)
    publish_time = fields.DateTime(required=True)


class BookResource(MethodResource, Resource):
    @doc(description='Get a book by book id', tags=['Book Requests'])
    def get(self, book_id: int):
        book = BookService().get_book_by_id(book_id)
        if book:
            return book.serialize(), 200
        else:
            return {'error': f'Book(book_id:{book_id}) not found'}, 404

    @doc(description='Update a book by book id', tags=['Book Requests'])
    @use_kwargs(TokenSchema, location='headers')
    @use_kwargs(BookRequestSchema, location='json')
    @token_required()
    def put(self, book_id: int, **kwargs):
        try:
            title = kwargs.get('title', None)
            author = kwargs.get('author', None)
            publish_time = kwargs.get('publish_time', None)
            book = Book(id=book_id, title=title, author=author, publish_time=publish_time)

            book = BookService().update_book(book)
            return book.serialize(), 200
        except Exception as e:
            return {'error': f'{e}'}, 500


class BookListResource(Resource):
    def get(self):
        book_list = BookService().get_all_books()
        return [book.serialize() for book in book_list]

    @token_required()
    def post(self):
        try:
            request_json = request.json
            if request_json:
                title = request_json.get('title', None)
                author = request_json.get('author', None)
                publish_time_str = request_json.get('publish_time', None)
                publish_time = datetime.fromisoformat(publish_time_str) if publish_time_str else None

                if title and author and publish_time:
                    book = Book(title=title, author=author, publish_time=publish_time)
                    book = BookService().create_book(book)
                    return book.serialize(), 200
                else:
                    return {'error': 'Invalid request, please provide complete book info'}, 400
            else:
                return {'error': 'Invalid request, please provide book info as json'}, 400
        except Exception as e:
            return {'error': f'{e}'}, 500


api.add_resource(BookResource, '/books/<int:book_id>')
api.add_resource(BookListResource, '/books')

docs.register(BookResource)


@app.route('/swagger.yaml')
def generate_swagger_yaml():
    yaml_doc = docs.spec.to_yaml()
    return Response(yaml_doc, mimetype='text/yaml')

response 内容

  • 同样，需要为 response 定义一个 schema 类 BookSchema,由于这个响应结果是对应的Book实体类；
  • 所以可以使用一个SQLAlchemyAutoSchema来实现返回数据库对应的实体类对象作为 response 结果；
  • 当然如果继续使用上面 RequestSchema 的形式是依旧可以实现的
1
2
3
4
class BookSchema(SQLAlchemyAutoSchema):
    class Meta:
        model = Book
        load_instance = True
1
2
3
4
5
6
7
8
9
10
11
12
flask==3.0.0
flask-restful==0.3.10
mysqlclient==2.1.1
SQLAlchemy==2.0.23
flask-SQLAlchemy==3.1.1
PyJWT==2.8.0
apispec==6.3.1
flask_apispec==0.11.4
PyYAML==6.0.1
flask_restful_swagger==0.20.2
marshmallow
marshmallow_sqlalchemy
  • 在 api 中使用这个类的时候，需要使用@marshal_with来使用上面的BookSchema
    • 使用@marshal_with 就可以直接返回实体类对象了，不需要再自己实现 serialized 序列化方法了
    • 同时，在 swagger 里面也会出现 response 的结果
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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
# -*- coding:utf-8 -*-
# Author: Zachary
from datetime import datetime

import jwt
from flask import request, Response
from flask_apispec import MethodResource, doc, use_kwargs, marshal_with
from flask_restful import Resource
from marshmallow import Schema, fields
from marshmallow_sqlalchemy import SQLAlchemyAutoSchema

from common.api_tools import token_required
from common.constants import LOGIN_SECRET
from models.book import Book
from resources import api, docs, app
from services.book_service import BookService


class TokenSchema(Schema):
    token = fields.String(required=True)


class BookRequestSchema(Schema):
    title = fields.String(required=True)
    author = fields.String(requird=True)
    publish_time = fields.DateTime(required=True)


class BookSchema(SQLAlchemyAutoSchema):
    class Meta:
        model = Book
        load_instance = True


class BookResource(MethodResource, Resource):
    @doc(description='Get a book by book id', tags=['Book Requests'])
    @marshal_with(BookSchema, 200)
    def get(self, book_id: int):
        book = BookService().get_book_by_id(book_id)
        if book:
            return book, 200
        else:
            return {'error': f'Book(book_id:{book_id}) not found'}, 404

    @doc(description='Update a book by book id', tags=['Book Requests'])
    @use_kwargs(TokenSchema, location='headers')
    @use_kwargs(BookRequestSchema, location='json')
    @marshal_with(BookSchema, 200)
    @token_required()
    def put(self, book_id: int, **kwargs):
        try:
            title = kwargs.get('title', None)
            author = kwargs.get('author', None)
            publish_time = kwargs.get('publish_time', None)
            book = Book(id=book_id, title=title, author=author, publish_time=publish_time)

            book = BookService().update_book(book)
            return book, 200
        except Exception as e:
            return {'error': f'{e}'}, 500


class BookListResource(Resource):
    def get(self):
        book_list = BookService().get_all_books()
        return [book.serialize() for book in book_list]

    @token_required()
    def post(self):
        try:
            request_json = request.json
            if request_json:
                title = request_json.get('title', None)
                author = request_json.get('author', None)
                publish_time_str = request_json.get('publish_time', None)
                publish_time = datetime.fromisoformat(publish_time_str) if publish_time_str else None

                if title and author and publish_time:
                    book = Book(title=title, author=author, publish_time=publish_time)
                    book = BookService().create_book(book)
                    return book.serialize(), 200
                else:
                    return {'error': 'Invalid request, please provide complete book info'}, 400
            else:
                return {'error': 'Invalid request, please provide book info as json'}, 400
        except Exception as e:
            return {'error': f'{e}'}, 500


api.add_resource(BookResource, '/books/<int:book_id>')
api.add_resource(BookListResource, '/books')

docs.register(BookResource)


@app.route('/swagger.yaml')
def generate_swagger_yaml():
    yaml_doc = docs.spec.to_yaml()
    return Response(yaml_doc, mimetype='text/yaml')

更新: 2024-02-10 23:22:16
原文: https://www.yuque.com/zacharyblock/cx2om6/ugubtklpuczalctm