介绍
在本教程中,我们将指导您完成使用 Flask 和 Python 创建 RESTful API 的过程。Flask 是一种流行的微框架,用于在 Python 中构建 Web 应用程序,而 RESTful API 是一种用于创建可扩展和可维护的 Web 服务的标准化架构。
先决条件
在开始之前,请确保您的系统上安装了以下软件:
- Python 3.x
pip
(Python 包安装程序)
1. 安装 Flask 和 Flask-RESTful
在安装 Flask 及其依赖项之前,让我们创建一个虚拟环境并激活它:
$ python -m venv venv
$ source venv/bin/activate # activate it on Linux/macOS
$ \venv\Scripts\activate # activate on Windows
现在我们需要使用 pip
安装 Flask 和 Flask-RESTful。打开终端或命令提示符并运行以下命令:
$ pip install Flask Flask-RESTful
2. 创建基本 Flask 应用程序
创建一个名为 app.py
的新文件,并添加以下代码:
from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello():
return "Hello, World!"
if __name__ == '__main__':
app.run(debug=True)
以上代码使用单个路由,设置基本的 Flask 应用程序。保存文件并使用以下命令运行应用程序:
$ python app.py
打开浏览器并访问 http://127.0.0.1:5000/。您会看到“Hello, World!”。
3. 使用 Flask-RESTful 创建 RESTful API
现在我们有了一个基本的 Flask 应用程序,我们可以开始构建我们的 RESTful API。修改 app.py
以包含以下代码:
from flask import Flask, request
from flask_restful import Resource, Api
app = Flask(__name__)
api = Api(app)
class HelloWorld(Resource):
def get(self):
return {'message': 'Hello, World!'}
api.add_resource(HelloWorld, '/api')
if __name__ == '__main__':
app.run(debug=True)
在这里,我们创建了一个继承自 Resource
的 HelloWorld
资源,并将其添加到 API 的 /api
终结点下。get()
方法返回一个包含消息的 JSON 对象。
4. 测试 API
保存对 app.py
所做的更改,然后重新启动应用程序。
要测试 API,您可以使用 curl
、Postman 甚至浏览器等工具。
打开您的浏览器并访问 http://127.0.0.1:5000/api。应看到以下 JSON 响应:
{
"message": "Hello, World!"
}
5. 扩展 API
现在,让我们扩展我们的 API 以处理更复杂的操作。
假设我们要创建一个用于管理任务列表的 API。使用以下代码更新 app.py
:
from flask import Flask, request
from flask_restful import Resource, Api
app = Flask(__name__)
api = Api(app)
tasks = []
class TaskList(Resource):
def get(self):
return {'tasks': tasks}
def post(self):
task = request.get_json()
tasks.append(task)
return {'tasks': tasks}
api.add_resource(TaskList, '/tasks')
if __name__ == '__main__':
app.run(debug=True)
现在,我们有一个 TaskList
资源,它有两个方法:get()
用于检索任务列表,post()
用于添加新任务。任务列表作为简单的 Python 列表在内存中维护。
常用的动作有:GET()
、POST()
、PUT()
、DELETE()
,分别对应,读取、添加、更新、删除。
6.关联数据库 (SQLite)
现在,我们将用 SQLite 数据库替换内存中列表以持久化任务。
SQLite 是一个轻量级、无服务器、独立的数据库,易于设置和使用。当然,你也可以使用其他数据库。
首先,安装 Flask-SQLAlchemy 扩展,以便更轻松地使用 SQLAlchemy 和 SQLite:
$ pip install Flask-SQLAlchemy
使用以下代码更新 app.py
:
from flask import Flask, request
from flask_restful import Resource, Api
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///tasks.db'
db = SQLAlchemy(app)
api = Api(app)
class Task(db.Model):
id = db.Column(db.Integer, primary_key=True)
description = db.Column(db.String(200), nullable=False)
def __repr__(self):
return f'Task {self.id}: {self.description}'
db.create_all()
class TaskList(Resource):
def get(self):
tasks = Task.query.all()
task_list = [{'id': task.id, 'description': task.description} for task in tasks]
return {'tasks': task_list}
def post(self):
task_data = request.get_json()
if not task_data:
return {'消息': '无有效的输入数据'}, 400
description = task_data.get('description')
if not description:
return {'消息': '无有效的描述数据'}, 400
new_task = Task(description=description)
db.session.add(new_task)
db.session.commit()
return {'消息': '已添加', 'task': {'id': new_task.id, 'description': new_task.description}}
api.add_resource(TaskList, '/tasks')
if __name__ == '__main__':
app.run(debug=True)
代码解释
在此段代码中,我们添加了 Flask-SQLAlchemy 包,并将其配置为使用名为 tasks.db
的 SQLite 数据库文件。我们创建了一个带有主键和描述的任务
模型。db.create_all()
函数创建数据库表(如果它们尚不存在)。
我们修改了 TaskList
资源以与 SQLite 数据库进行交互。get()
方法从数据库中检索所有任务,post()
方法向数据库添加一个新任务,并提交更改。
当您在没有有效 JSON 有效负载的情况下发出 POST 请求时,API 将返回一条带有 400 状态代码的错误消息,指示请求不正确。此外,该代码还会检查请求数据中是否存在描述字段,如果缺少描述字段,则返回错误消息。
重新启动应用程序并使用您喜欢的工具(例如,curl
、Postman 或 浏览器)测试 API。
这些任务现在应在会话之间保留,并存储在 tasks.db
SQLite 数据库文件中。
7. 将代码分成多个文件
在实际应用程序中,最好将代码分成多个文件,以保持干净有序的项目结构。
这种模块化使您的代码更易于理解、维护和扩展。
下面是一个示例,说明如何将本教程中的代码分成多个文件:
config.py
:存储应用程序的配置设置,例如数据库 URI:
SQLALCHEMY_DATABASE_URI = 'sqlite:///tasks.db'
models.py
:定义数据库模型和架构:
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
class Task(db.Model):
id = db.Column(db.Integer, primary_key=True)
description = db.Column(db.String(200), nullable=False) # nullable=False means that the column cannot be empty
def __repr__(self):
# This method is used to print the object.
return f'Task {self.id}: {self.description}'
resources.py
:定义 API 资源和逻辑:
from flask_restful import Resource
from flask import request
from models import Task, db
class TaskList(Resource):
def get(self):
# Get all the tasks from the database.
tasks = Task.query.all()
# Convert the tasks to JSON and return a response.
task_list = [{'id': task.id, 'description': task.description} for task in tasks]
return {'tasks': task_list}
def post(self):
# Get the JSON data from the request.
task_data = request.get_json()
# Check if the data is valid.
if not task_data:
return {'message': 'No input data provided'}, 400
description = task_data.get('description')
if not description:
return {'message': 'Description is required'}, 400
# Add the task to the database.
new_task = Task(description=description)
db.session.add(new_task)
# Commit the task to the database.
db.session.commit()
# Return a message to the user.
return {'message': 'Task added', 'task': {'id': new_task.id, 'description': new_task.description}}
app.py
:初始化应用程序、导入资源并定义 API 路由:
from flask import Flask
from flask_restful import Api
from models import db
import config
from resources import TaskList
# Create the Flask application and the Flask-RESTful API manager.
app = Flask(__name__)
app.config.from_object(config)
# Initialize the Flask-SQLAlchemy object.
db.init_app(app)
# Create the Flask-RESTful API manager.
api = Api(app)
# Create the endpoints.
api.add_resource(TaskList, '/tasks')
if __name__ == '__main__':
# Create the database tables.
with app.app_context():
db.create_all()
# Start the Flask development web server.
app.run(debug=True)
通过将代码分成不同的文件,您可以轻松地管理应用程序的组件并维护组织良好的项目结构。
此外,这种模块化方法还允许你重用代码并更有效地扩展应用程序。
运行应用程序后,这是我使用 Postman 执行的 POST 请求:
确保选择 POST 方法。前往“正文”选项卡,选择“原始”,然后从单选按钮右侧的下拉菜单中选择“JSON”。
执行两次 POST 请求后,下面是一个 GET 请求: