关于程序员:手把手带你入门-API-开发

3次阅读

共计 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 版本，本文以 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
)

咱们上个步骤中看到数据库将被称为 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 操作。

在 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 中的所有内容了

能够查看官网装置帮忙：

$ 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 作为接口测试工具：

拜访 http://127.0.0.1:8000/games。返回 200 OK，然而目前数据库中没有游戏，所以返回空 JSON 字符串：

因为游戏 id 是自增的，所以只须要往外面的服务器传入 name, price 和 rate：

服务器后盾显示如下：

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

查看咱们以后的所有的游戏内容：

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

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

测试胜利如下：

为了查看刚刚针对 id 为 2 的游戏是否更名胜利，咱们能够传入http://127.0.0.1:8000/game/2 门路：

能够看到，PUT 和以后的 GET 申请都是 OK 的。

最初，就来到咱们的删除环节，把 id 为 3 的游戏内容删除：

能够看到删除胜利，咱们的 API 测试是齐全 OK 的

最初，又到了本文须要做总结的时候了。

本篇教程利用 Python、SQLite3、Flask 实现了一个简略的 游戏 Rest API 性能，通过在 Flask 中解决传入的 HTTP 的四大申请：GET、POST、PUT、DELETE，实现了最根本的增删改查性能。

最初利用 Apifox 作为咱们的接口测试工具，残缺的体验了一个繁难 API 开发的流程。

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

正文完

程序员

发表至：程序员

2022-07-01

0

关于程序员:面试重点建立Java并发知识体系含工具全图鉴

关于程序员:考PMP需要做什么准备考试内容是什么

关于程序员:文献阅读在整个小鼠大脑中细胞类型的高分辨率转录组和空间图谱

关于程序员:elixir-0075-打工人的总结之编程的术与道

关于jdbc:JDBC-在性能测试中的应用

关于程序员:手把手带你入门-API-开发

引言

装置 SQLite

创立数据库

创立表

创立 db.py

创立 game_controller.py

创立 main.py

装置 flask

启动程序并测试

GET 获取所有游戏测试

POST 新增一个游戏测试

PUT 批改游戏内容测试

GET 获取单个游戏内容测试

DELETE 删除游戏测试

总结

站内搜索