使用 SQLite 构建 Flask REST API | 编程教程

介绍

在这个项目中，我们将学习如何使用 SQLite 作为数据库在 Flask 中构建一个 REST API。我们将首先创建一个示例 SQLite 数据库并用一些数据填充它。然后，我们将构建一个 Flask 应用程序，该应用程序具有用于对 SQLite 数据库中的数据执行 CRUD（创建、读取、更新、删除）操作的端点。

👀 预览

Flask SQLite API 预览

🎯 任务

在这个项目中，你将学习：

如何创建一个以 SQLite 作为数据库后端的 Flask 应用程序
如何创建一个示例 SQLite 数据库和表
如何用示例数据填充数据库表
如何实现端点以检索所有用户、通过 ID 检索特定用户、创建新用户、更新现有用户以及删除用户
如何使用 cURL 或浏览器测试 REST API 端点

🏆 成果

完成这个项目后，你将能够：

构建一个以 SQLite 作为数据库的 Flask 应用程序
使用 Flask 对 SQLite 数据库中的数据执行 CRUD 操作
为 Flask 应用程序实现并测试 REST API 端点

创建项目文件

让我们从为我们的 Flask 应用程序创建必要的项目文件开始。

使用 cd 命令进入 ~/project 目录。
在项目目录中，创建一个名为 app.py 的新 Python 文件。
在同一目录中创建一个名为 create_database.py 的新文件。

cd ~/project
touch app.py create_database.py

你的项目结构应该如下所示：

project/
├── app.py
└── create_database.py

创建一个示例 SQLite 数据库

接下来，让我们使用 Python 和 SQLite 库创建一个 SQLite 数据库并用示例数据填充它。

在 create_database.py 文件中，添加以下代码以导入所需的模块：

import sqlite3

添加以下代码以创建数据库和表：

## 创建到数据库的连接
conn = sqlite3.connect('example.db')

## 创建一个游标对象以执行 SQL 语句
cursor = conn.cursor()

## 创建一个名为 "users" 的表
cursor.execute('''CREATE TABLE IF NOT EXISTS users (
                    id INTEGER PRIMARY KEY,
                    name TEXT,
                    email TEXT
                )''')

接下来，让我们向表中插入一些示例数据：

## 向 "users" 表中插入示例数据
users = [
    ('John Doe', 'johndoe@example.com'),
    ('Jane Smith', 'janesmith@example.com'),
    ('Bob Johnson', 'bobjohnson@example.com')
]
cursor.executemany('INSERT INTO users (name, email) VALUES (?,?)', users)

## 将更改提交到数据库
conn.commit()

## 关闭游标和数据库连接
cursor.close()
conn.close()

cd ~/project
python create_database.py

运行此代码后，你将拥有一个名为 example.db 的 SQLite 数据库文件，其中的 "users" 表包含示例数据。这个数据库将在我们的 Flask 应用程序中用于检索和操作数据。

在 Flask 中构建 REST API

现在，让我们使用之前创建的 SQLite 数据库 example.db 在 Flask 中构建一个 REST API。

在 app.py 文件中，添加以下代码以导入所需的模块：

from flask import Flask, jsonify, request
import sqlite3

添加以下代码以创建 Flask 应用程序：

## 创建 Flask 应用程序
app = Flask(__name__)

现在，让我们为我们的 REST API 定义端点：

检索所有用户的端点

## 检索所有用户的端点
@app.route('/users', methods=['GET'])
def get_users():
    ## 连接到数据库
    conn = sqlite3.connect('example.db')
    cursor = conn.cursor()

    ## 执行 SQL 查询以检索所有用户
    cursor.execute('SELECT * FROM users')
    users = cursor.fetchall()

    ## 关闭游标和数据库连接
    cursor.close()
    conn.close()

    ## 将用户作为 JSON 返回
    return jsonify(users)

在这段代码中，我们首先使用 connect() 方法连接到 SQLite 数据库。然后，我们使用 cursor() 方法创建一个游标对象来执行 SQL 语句。接下来，我们使用 execute() 方法执行 SQL 查询以从 "users" 表中检索所有用户。最后，我们使用 close() 方法关闭游标和数据库连接，并使用 jsonify() 方法将用户作为 JSON 返回。

通过 ID 检索特定用户的端点

## 通过 ID 检索特定用户的端点
@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    ## 连接到数据库
    conn = sqlite3.connect('example.db')
    cursor = conn.cursor()

    ## 执行 SQL 查询以按 ID 检索用户
    cursor.execute('SELECT * FROM users WHERE id =?', (user_id,))
    user = cursor.fetchone()

    ## 关闭游标和数据库连接
    cursor.close()
    conn.close()

    ## 检查用户是否存在
    if user:
        return jsonify(user)
    else:
        return jsonify({'message': '用户未找到'}), 404

此端点与上一个类似，不同之处在于它按 ID 检索特定用户而不是所有用户。我们使用 fetchone() 方法从 "users" 表中检索单个用户。

创建新用户的端点

## 创建新用户的端点
@app.route('/users', methods=['POST'])
def create_user():
    ## 从请求体中获取用户数据
    data = request.get_json()
    name = data['name']
    email = data['email']

    ## 连接到数据库
    conn = sqlite3.connect('example.db')
    cursor = conn.cursor()

    ## 执行 SQL 查询以插入新用户
    cursor.execute('INSERT INTO users (name, email) VALUES (?,?)', (name, email))
    conn.commit()

    ## 关闭游标和数据库连接
    cursor.close()
    conn.close()

    ## 返回成功消息
    return jsonify({'message': '用户创建成功'})

此端点用于在 "users" 表中创建新用户。我们使用 get_json() 方法从请求体中获取用户数据。然后，我们使用 execute() 方法将新用户插入到 "users" 表中。

更新现有用户的端点

## 更新现有用户的端点
@app.route('/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    ## 从请求体中获取更新后的用户数据
    data = request.get_json()
    name = data['name']
    email = data['email']

    ## 连接到数据库
    conn = sqlite3.connect('example.db')
    cursor = conn.cursor()

    ## 执行 SQL 查询以更新用户
    cursor.execute('UPDATE users SET name =?, email =? WHERE id =?', (name, email, user_id))
    conn.commit()

    ## 关闭游标和数据库连接
    cursor.close()
    conn.close()

    ## 返回成功消息
    return jsonify({'message': '用户更新成功'})

此端点用于在 "users" 表中更新现有用户。我们使用 get_json() 方法从请求体中获取更新后的用户数据。然后，我们使用 execute() 方法在 "users" 表中更新用户。

删除用户的端点

## 删除用户的端点
@app.route('/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    ## 连接到数据库
    conn = sqlite3.connect('example.db')
    cursor = conn.cursor()

    ## 执行 SQL 查询以删除用户
    cursor.execute('DELETE FROM users WHERE id =?', (user_id,))
    conn.commit()

    ## 关闭游标和数据库连接
    cursor.close()
    conn.close()

    ## 返回成功消息
    return jsonify({'message': '用户删除成功'})

此端点用于从 "users" 表中删除用户。我们使用 execute() 方法从 "users" 表中删除用户。

最后，让我们添加以下代码来运行 Flask 应用程序：

if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0', port=8080)

运行 Flask 应用程序

要运行该项目并启动 Flask 应用程序，请打开终端或命令提示符，导航到项目目录，然后执行以下命令：

cd ~/project
python app.py

你应该会看到类似于以下的输出：

* 正在提供 Flask 应用 'app'（延迟加载）
* 环境：生产
警告：这是一个开发服务器。请勿在生产部署中使用它。
请改用生产 WSGI 服务器。
* 调试模式：开启
* 正在运行于 http://0.0.0.0:8080/（按 CTRL+C 退出）

这意味着 Flask 应用程序正在你本地的机器上运行。现在，你可以使用诸如 cURL 之类的工具或像 Postman 这样的 API 测试工具来测试 REST API 端点。

测试 REST API 端点

打开一个新的终端或命令提示符。
使用以下 cURL 命令测试每个端点：

检索所有用户：

curl http://localhost:8080/users

按 ID 检索特定用户（将 <user_id> 替换为实际的用户 ID）：

curl http://localhost:8080/users/<user_id>

创建新用户：

curl -X POST -H "Content-Type: application/json" -d '{"name":"John Doe","email":"johndoe@example.com"}' http://localhost:8080/users

更新现有用户（将 <user_id> 替换为实际的用户 ID）：

curl -X PUT -H "Content-Type: application/json" -d '{"name":"Updated Name","email":"updated@example.com"}' http://localhost:8080/users/<user_id>

删除用户（将 <user_id> 替换为实际的用户 ID）：

curl -X DELETE http://localhost:8080/users/<user_id>

在测试相应端点时，请确保将 <user_id> 替换为数据库中实际的用户 ID。

通过运行这些 cURL 命令，你可以测试 REST API 端点的功能并验证它们是否正常工作。

或者，你可以切换到 Web 8080 标签页，使用浏览器测试 REST API 端点。

REST API 测试界面

通过这些测试步骤，你可以确保 Flask 应用程序及其 REST API 端点按预期运行。

总结

在这个项目中，我们学习了如何使用 SQLite 作为数据库在 Flask 中构建一个 REST API。我们创建了一个示例 SQLite 数据库并用数据填充它。然后，我们构建了一个 Flask 应用程序，其端点用于从数据库中检索、创建、更新和删除用户。按照逐步说明操作，你现在应该拥有一个功能正常的 Flask 应用程序，它提供了一个 REST API 来与 SQLite 数据库进行交互。