手把手带你入门 API 开发程序员

引言
文章图片

在本文中，您将学习如何使用 Flask、SQLite 3（轻易数据库）和 JSON 创建用于数据通信的 REST API。
本文使用 4 个最常用的 HTTP 动词：GET、POST、PUT 和 DELETE，对应数据库的 CRUD 操作。
比如管理的是一个游戏数据库 games.db，其中包含名称（name）、价格（price) 和等级（rate）。
我们还将通过使用 Flask 创建的 API 公开几个操作：

获取所有游戏
创建一个新游戏
更新游戏
删除游戏
通过 ID 获取游戏

首先，我们将使用 Python 创建与数据库相关的 CRUD，然后我们将在 API 中使用 Flask 公开所有这些函数，编码格式为 JSON。
安装 SQLite

点击此处，下载你的系统对应的 SQLite 版本，本文以 Windows 为例：

文章图片

下载后将这两个压缩包内的文件解压到 C 盘的某个目录下：

文章图片

查看 SQLite3 版本：

λ sqlite3 SQLite version 3.38.5 2022-05-06 15:25:27 Enter ".help" for usage hints. Connected to a transient in-memory database. Use ".open FILENAME" to reopen on a persistent database. sqlite>

创建数据库使用 sqlite3 databaseName.db 命令来创建一个 SQLite 数据库，本文创建一个 games.db 数据库：

λ sqlite3 games.db SQLite version 3.38.5 2022-05-06 15:25:27 Enter ".help" for usage hints. sqlite> .databases main: C:\Program Files\cmder\games.db r/w

创建表

CREATE TABLE IF NOT EXISTS games( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, price REAL NOT NULL, rate INTEGER NOT NULL )

创建 db.py 我们上个步骤中看到数据库将被称为 games.db 。新建 Python 的 SQLite3 连接文件 db.py：

import sqlite3DATABASE_NAME = "games.db"# 获取数据库连接 def get_db(): conn = sqlite3.connect(DATABASE_NAME) return conn# 创建数据库表 def create_tables(): tables = [ """ CREATE TABLE IF NOT EXISTS games ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, price REAL NOT NULL, rate INTEGER NOT NULL ) """ ]db = get_db() cursor = db.cursor() for table in tables: cursor.execute(table)

中，此外，我们有两个功能：

其中之一是 get_db()：用于获得数据库连接
另一个功能 create_tables()：是当 games 数据库表不存在的情况下创建数据库表。

现在我们已经定义了数据库，让我们看看使用 SQLite3 数据库的游戏的 CRUD 操作。
创建 game_controller.py 在 API 中公开数据库之前，我们将创建一个游戏控制器，它将负责保存、更新、删除和获取游戏数据的所有操作。
所有这些函数都在一个名为 game_controller.py 的文件中，它看起来像这样：

from db import get_dbdef insert_game(name, price, rate): db = get_db() cursor = db.cursor() statement = "INSERT INTO games(name, price, rate) VALUES (?, ?, ?)" cursor.execute(statement, [name, price, rate]) db.commit() return Truedef update_game(id, name, price, rate): db = get_db() cursor = db.cursor() statement = "UPDATE games SET name = ?, price = ?, rate = ? WHERE id = ?" cursor.execute(statement, [name, price, rate, id]) db.commit() return Truedef delete_game(id): db = get_db() cursor = db.cursor() statement = "DELETE FROM games WHERE id = ?" cursor.execute(statement, [id]) db.commit() return Truedef get_by_id(id): db = get_db() cursor = db.cursor() statement = "SELECT id, name, price, rate FROM games WHERE id = ?" cursor.execute(statement, [id]) return cursor.fetchone()def get_games(): db = get_db() cursor = db.cursor() query = "SELECT id, name, price, rate FROM games" cursor.execute(query) return cursor.fetchall()

在文件中，我们看到了几个函数。 insert_game 函数接收游戏数据并将其插入数据库（INSERT）；所有这些都使用准备好的语句来避免我们使用 Python 和 Flask 创建的这个 API 中的 SQL 注入。
我们还看到其他方法，例如 update_game 执行 UPDATE 操作以更新游戏，delete_game 从其 id 删除游戏 (DELETE)，get_by_id 从其 id 返回游戏（使用 SELECT 操作）。
最后我们看看返回所有现有游戏的 get_games 函数。
请注意，所有函数都使用数据库和游标来执行所有操作。现在我们有了数据库操作的 CRUD，是时候用 Flask 公开 API 中的所有内容了
创建 main.py

文章图片

安装 flask

文章图片

可以查看官方安装帮助：

$ pip install Flask

我们在 API 中做的第一件事是创建 Flask 应用程序并导入游戏控制器。我们还从数据库中导入了一个函数，因为我们需要在启动应用程序时创建表:

from flask import Flask, jsonify, request import game_controller from db import create_tablesapp = Flask(__name__)

然后使用 GET、PUT、POST 和 DELETE http 动词定义路由：

@app.route('/games', methods=["GET"]) def get_games(): games = game_controller.get_games() return jsonify(games)@app.route("/game", methods=["POST"]) def insert_game(): game_details = request.get_json() name = game_details["name"] price = game_details["price"] rate = game_details["rate"] result = game_controller.insert_game(name, price, rate) return jsonify(result)@app.route("/game", methods=["PUT"]) def update_game(): game_details = request.get_json() id = game_details["id"] name = game_details["name"] price = game_details["price"] rate = game_details["rate"] result = game_controller.update_game(id, name, price, rate) return jsonify(result)@app.route("/game/", methods=["DELETE"]) def delete_game(id): result = game_controller.delete_game(id) return jsonify(result)@app.route("/game/", methods=["GET"]) def get_game_by_id(id): game = game_controller.get_by_id(id) return jsonify(game)

每条路由都对应了我们之前创建的游戏控制器功能，从而与 SQLite3 数据库进行交互。

在更新和插入游戏时，我们使用 get_json 阅读了请求的 JSON ，然后访问字典:

game_details = request.get_json()

在删除或通过 ID 获取的情况下，我们从路由中读取 id 变量作为并在方法中接收它:

game = game_controller.get_by_id(id)

此 Python API 通过 JSON 进行通信，因此所有响应都是根据 jsonify 函数返回的内容进行的:

return jsonify(result)

最后，我们创建 Flask 应用程序来启动服务器并监听请求：

if __name__ == "__main__": create_tables() """ Here you can change debug and port Remember that, in order to make this API functional, you must set debug in False """ app.run(host='0.0.0.0', port=8000, debug=False)

至此，我们完整的 main.py 函数如下：

from flask import Flask, jsonify, request, json import game_controller from db import create_tablesapp = Flask(__name__)@app.route('/games', methods=["GET"]) def get_games(): games = game_controller.get_games() return jsonify(games)@app.route("/game", methods=["POST"]) def insert_game():game_details = json.loads(request.get_data()) name = game_details["name"] price = game_details["price"] rate = game_details["rate"] result = game_controller.insert_game(name, price, rate) return jsonify(result)@app.route("/game", methods=["PUT"]) def update_game(): game_details = request.get_json() id = game_details["id"] name = game_details["name"] price = game_details["price"] rate = game_details["rate"] result = game_controller.update_game(id, name, price, rate) return jsonify(result)@app.route("/game/", methods=["DELETE"]) def delete_game(id): result = game_controller.delete_game(id) return jsonify(result)@app.route("/game/", methods=["GET"]) def get_game_by_id(id): game = game_controller.get_by_id(id) return jsonify(game)""" Enable CORS. Disable it if you don't need CORS """ @app.after_request def after_request(response): response.headers["Access-Control-Allow-Origin"] = "*" # <- You can change "*" for a domain for example "http://localhost" response.headers["Access-Control-Allow-Credentials"] = "true" response.headers["Access-Control-Allow-Methods"] = "POST, GET, OPTIONS, PUT, DELETE" response.headers["Access-Control-Allow-Headers"] = "Accept, Content-Type, Content-Length, Accept-Encoding, X-CSRF-Token, Authorization" return responseif __name__ == "__main__": create_tables() """ Here you can change debug and port Remember that, in order to make this API functional, you must set debug in False """ app.run(host='0.0.0.0', port=8000, debug=False)

值得注意的是，这里我们添加 CORS，如果您要从与 API 本身不同的域使用此 API，则需要启用 CORS。只需将以下代码片段添加到 API（在存储库中，您会发现已添加的代码，您可以根据需要将其删除）：

""" Enable CORS. Disable it if you don't need CORS """ @app.after_request def after_request(response): response.headers["Access-Control-Allow-Origin"] = "*" # <- You can change "*" for a domain for example "http://localhost" response.headers["Access-Control-Allow-Credentials"] = "true" response.headers["Access-Control-Allow-Methods"] = "POST, GET, OPTIONS, PUT, DELETE" response.headers["Access-Control-Allow-Headers"] = "Accept, Content-Type, Content-Length, Accept-Encoding, X-CSRF-Token, Authorization" return response

启动程序并测试使用 python3 main.py，运行我们的服务器和 API ：

* Serving Flask app 'main' (lazy loading) * Environment: production WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead. * Debug mode: off * Running on all addresses (0.0.0.0) WARNING: This is a development server. Do not use it in a production deployment. * Running on http://127.0.0.1:8000 * Running on http://10.26.4.188:8000 (Press CTRL+C to quit)

为了方便本文创建的所有 API 的测试，使用到 Apifox 作为接口测试工具：

文章图片

GET 获取所有游戏测试
访问 http://127.0.0.1:8000/games。返回 200 OK，但是目前数据库中没有游戏，所以返回空 JSON 字符串：

文章图片

POST 新增一个游戏测试
因为游戏 id 是自增的，所以只需要往外面的服务器传入 name, price 和 rate：

文章图片

服务器后台显示如下：

127.0.0.1 - - [16/Jun/2022 14:51:05] "POST /game HTTP/1.1" 200 -

PUT 修改游戏内容测试
查看我们当前的所有的游戏内容：

文章图片

然后将 LOL 的名字改为英雄联盟，此时我们需要传入四个参数：id，name, price, rate，因为要根据 id 来找到我们想要更改的那个游戏：

{ "id": 2, "name": "英雄联盟", "price": 1, "rate": 99 }

测试成功如下：

文章图片

GET 获取单个游戏内容测试
为了检查刚刚针对 id 为 2 的游戏是否更名成功，我们可以传入http://127.0.0.1:8000/game/2 路径：

文章图片

可以看到，PUT 和当前的 GET 请求都是 OK 的。
DELETE 删除游戏测试
最后，就来到我们的删除环节，把 id 为 3 的游戏内容删除：

文章图片

可以看到删除成功，我们的 API 测试是完全 OK 的

文章图片

总结最后，又到了本文需要做总结的时候了。
本篇教程利用 Python、SQLite3、Flask 实现了一个简单的游戏 Rest API 功能，通过在 Flask 中处理传入的 HTTP 的四大请求：GET、POST、PUT、DELETE，实现了最基本的增删改查功能。
最后利用 Apifox 作为我们的接口测试工具，完整的体验了一个简易 API 开发的流程。
【手把手带你入门 API 开发】如果文章对你有用的话，点个赞再离开吧，下一篇文章再见。