共计 7420 个字符,预计需要花费 19 分钟才能阅读完成。
引言
在本文中,您将学习如何应用 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 sqlite3
DATABASE_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_db
def 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 True
def 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 True
def delete_game(id):
db = get_db()
cursor = db.cursor()
statement = "DELETE FROM games WHERE id = ?"
cursor.execute(statement, [id])
db.commit()
return True
def 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_tables
app = 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, json
import game_controller
from db import create_tables
app = 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_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
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)
值得注意的是,这里咱们增加 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 开发的流程。
如果文章对你有用的话,点个赞再来到吧,下一篇文章再见。