关于程序员:手把手带你入门-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

1. 点击此处，下载你的零碎对应的 SQLite 版本，本文以 Windows 为例：

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

1. 并将该目录增加至环境变量：

1. 查看 SQLite3 版本：

 sqlite3SQLite version 3.38.5 2022-05-06 15:25:27Enter ".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.dbSQLite version 3.38.5 2022-05-06 15:25:27Enter ".help" for usage hints.sqlite> .databasesmain: 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, requestimport game_controllerfrom 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/<id>", methods=["DELETE"])def delete_game(id):    result = game_controller.delete_game(id)    return jsonify(result)@app.route("/game/<id>", 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 变量作为 <variable> 并在办法中接管它:

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, jsonimport game_controllerfrom 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/<id>", methods=["DELETE"])def delete_game(id):    result = game_controller.delete_game(id)    return jsonify(result)@app.route("/game/<id>", 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_requestdef 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_requestdef 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 开发的流程。

如果文章对你有用的话，点个赞再来到吧，下一篇文章再见。