当前位置：首页 > article >正文

Python Flask 开发用于访问数据库的 REST API

article 2025/3/12 13:18:29

Python Flask 开发用于访问数据库的 REST API

Python Flask 开发用于访问数据库的 REST API

Python Flask 开发用于访问数据库的 REST API

为了说明如何使用 Flask 和 Flask-RESTful 扩展构建 REST API，我们将修改之前网页应用程序，并使用 REST 架构风格访问 Student 对象（Resource对象）。我们希望在请求正文中为 PUT 和 POST 方法发送参数，而 API 将以 JSON 格式发回响应。带有 REST API 接口的修订版代码如下所示：

from flask_sqlalchemy import SQLAlchemy
from flask import Flask
from flask_restful import Api, Resource, reqparse

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

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///student.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

在前面的代码片段中，我们首先初始化了 Flask 应用程序和数据库实例。下一步，我们使用 Flask 实例创建了一个 API 实例。我们通过 api = Api(app) 语句实现了这一点。这个 API 实例是开发其余 API 应用程序的关键，我们将使用它。

接下来，我们需要配置 reqparse 实例，注册我们希望从 HTTP 请求中解析的参数。在我们的代码示例中，我们注册了两个字符串类型的参数：name 和 grade，如以下代码片段所示：

parser = reqparse.RequestParser()
parser.add_argument('name', type=str)
parser.add_argument('grade', type=str)

下一步是创建 Student 模型对象，这与我们前面写过的代码中的操作相同，只是我们将添加一个 serialize 方法，将对象转换为 JSON 格式。这是将 JSON 响应序列化后再发送回发送给 API 客户端之前序列化 JSON 响应的重要步骤。还有其他解决方案可以实现同样的目的，但出于简单的原因，我们选择了此选项。创建 Student 对象的精确示例代码如下：

class Student(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), nullable=False)
    grade = db.Column(db.String(20), nullable=True)

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

接下来，我们创建两个 Resource 对象，用来访问学生数据库对象。有两个 Resource 对象，分别是 StudentDao 和 StudentListDao，下面是对做它们的描述：

StudentDao：它为每一个单独的资源实例提供了像 get 和 delete 这样的方法，这些方法会被映射到 HTTP 协议的 GET 和 DELETGE 方法；
StudentListDao：它提供了像 get 和 post 这样的方法。添加 get 方法是为了使用 HTTP 的 GET 方法提供所有的 Student 类型的资源的列表。添加 post 方法是为了使用 HTTP 的 POST 方法添加一个新资源对象。这是一种典型的设计模式，用于实现网页资源的 CRUD 功能。

至于为 StudentDao 和 StudentListDao 类实现的方法，我们在一条语句中返回状态代码和对象本身。这是 Flask-RESTful 扩展提供的便利。下面是这两个类的代码：

class StudentDao(Resource):
    def get(self, student_id):
        student = Student.query.filter_by(id=student_id).first_or_404(description='没有 ID 是 {} 的学生'.format(student_id))
        return student.serialize()

    def delete(self, student_id):
        student = Student.query.filter_by(id=student_id).first_or_404(description='没有 ID 是 {} 的学生'.format(student_id))
        db.session.delete(student)
        db.commit()
        return '', 204

class StudentListDao(Resource):
    def get(self):
        students = Student.query.all()
        return [Student.serialize(student) for student in students]

    def post(self):
        args = parser.parse_args()
        name - =args['name']
        grade - =args['grade']
        student = Student(name=name, grade=grade)
        db.session.add(student)
        db.session.commit()
        return student, 200

对于 StudentListDao 类的 post 方法，我们使用 reqparse 解析器从请求中提取了姓名和成绩参数。POST 方法的其他实现与前面应用程序示例中的相同。

在示例 API 应用程序的后面两行中，我们将 URL 映射到了资源对象。所有指向 /students/<student_id> 的请求都将重定向到 StudentDao 资源类。所有指向 /students 的请求都将重定向到 StudentListDao 资源类：

api.add_resource(StudentDao, '/student/<student_id>')
api.add_resource(StudentListDao, '/student/list')

对于在本代码示例中，我们没有添加错误和异常处理，以保持代码简洁，便于讨论。但我们强烈建议在最终实现中加入错误和异常处理。但我们强烈建议在最终实现中加入这一点。

下面是完整的代码：

# 学生资源的 REST API 应用程序
from flask_sqlalchemy import SQLAlchemy
from flask import Flask
from flask_restful import Api, Resource, reqparse

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

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///student.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

parser = reqparse.RequestParser()
parser.add_argument('name', type=str)
parser.add_argument('grade', type=str)


class Student(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), nullable=False)
    grade = db.Column(db.String(20), nullable=True)

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

class StudentDao(Resource):
    def get(self, student_id):
        student = Student.query.filter_by(id=student_id).first_or_404(description='没有 ID 是 {} 的学生'.format(student_id))
        return student.serialize()

    def delete(self, student_id):
        student = Student.query.filter_by(id=student_id).first_or_404(description='没有 ID 是 {} 的学生'.format(student_id))
        db.session.delete(student)
        db.commit()
        return '', 204

class StudentListDao(Resource):
    def get(self):
        students = Student.query.all()
        return [Student.serialize(student) for student in students]

    def post(self):
        args = parser.parse_args()
        name - =args['name']
        grade - =args['grade']
        student = Student(name=name, grade=grade)
        db.session.add(student)
        db.session.commit()
        return student, 200

api.add_resource(StudentDao, '/student/<student_id>')
api.add_resource(StudentListDao, '/student/list')