SQLite3 与烧瓶 Web 应用程序 CRUD

本文将介绍使用 Python 的 SQLite3 数据库和烧瓶 Web 应用程序进行 CRUD 操作的示例。CRUD 是指创建（Create）、读取（Read）、更新（Update）和删除（Delete）数据的操作。

安装烧瓶和 SQLite3 库

在开始之前，确保您已经安装了 Python、烧瓶和 SQLite3 库。如果您还没有安装它们，请按照以下步骤安装：

安装 Python

Python 可以从官方网站下载并安装。

安装烧瓶

打开命令提示符或终端窗口，运行以下命令：

pip install Flask

安装 SQLite3 库

SQLite3 库通常已经包含在 Python 中。如果您需要最新版本的 SQLite3 库，请使用以下命令：

pip install pysqlite3

创建 SQLite3 数据库

在这个示例中，我们将使用 SQLite3 数据库来存储书籍信息。我们需要创建一个具有以下数据列的 book 表：

• id：书籍的唯一标识符
• title：书籍的标题
• author：书籍的作者
• isbn：书籍的 ISBN 号码

我们可以使用以下 SQL 命令创建 book 表：

CREATE TABLE book (
    id INTEGER PRIMARY KEY,
    title TEXT NOT NULL,
    author TEXT NOT NULL,
    isbn TEXT NOT NULL
);

我们可以使用 SQLite3 命令行工具或其他 SQLite3 可视化工具（如 SQLiteStudio）来创建数据库和表。或者，我们可以使用 Python 脚本来创建它们。在这个示例中，我们将使用 Python 脚本。

创建一个名为 create_db.py 的 Python 文件并插入以下代码：

import sqlite3

conn = sqlite3.connect('books.db')
c = conn.cursor()

c.execute('''
    CREATE TABLE book (
        id INTEGER PRIMARY KEY,
        title TEXT NOT NULL,
        author TEXT NOT NULL,
        isbn TEXT NOT NULL
    );
''')

conn.commit()
conn.close()

这个代码将创建一个名为 books.db 的 SQLite3 数据库文件和一个包含四个数据列的 book 表。

插入示例数据

现在，让我们向 book 表插入一些示例数据。插入数据的过程称为“创建”或“C”操作。

在我们的示例中，我们将插入三本书的信息。创建一个名为 insert_data.py 的 Python 文件并插入以下代码：

import sqlite3

conn = sqlite3.connect('books.db')
c = conn.cursor()

books = [
    ('The Great Gatsby', 'F. Scott Fitzgerald', '9780141182636'),
    ('To Kill a Mockingbird', 'Harper Lee', '9780446310789'),
    ('The Catcher in the Rye', 'J.D. Salinger', '9780316769488')
]

c.executemany('INSERT INTO book (title, author, isbn) VALUES (?, ?, ?)', books)

conn.commit()
conn.close()

这个代码将使用 executemany() 方法向 book 表插入三本书的信息。

创建烧瓶 Web 应用程序

现在我们已经有了一个 SQLite3 数据库和一些示例数据，让我们创建一个基于烧瓶的 Web 应用程序来进行 CRUD 操作。

创建 Flask 应用程序

在您的 Python 项目目录中，创建一个名为 app.py 的文件，并插入以下代码：

from flask import Flask, jsonify, request
import sqlite3

app = Flask(__name__)

@app.route('/')
def hello():
    return 'Hello, World!'

if __name__ == '__main__':
    app.run(debug=True)

这个代码创建了一个基本的 Flask 应用程序，并定义了一个路由 /，它将返回 “Hello, World!” 消息。

我们可以在命令提示符或终端窗口中运行 python app.py 命令来启动该应用程序，并在浏览器中访问 http://127.0.0.1:5000/ 来查看消息。

获取所有图书

让我们在我们的应用程序中创建一个路由来获取 book 表中包含的所有图书的信息。这是一个“读取”或“R”操作。

在 app.py 中插入以下代码：

@app.route('/books', methods=['GET'])
def get_all_books():
    conn = sqlite3.connect('books.db')
    c = conn.cursor()

    c.execute('SELECT * FROM book')

    books = []
    for row in c.fetchall():
        books.append({
            'id': row[0],
            'title': row[1],
            'author': row[2],
            'isbn': row[3]
        })

    conn.close()

    return jsonify(books)

这个代码定义了一个 /books 路由，并使用 GET HTTP 方法来获取 book 表中包含的所有图书的信息。它使用 SQLite3 的 connect() 和 cursor() 方法来连接到数据库并执行 SQL 查询。查询返回的结果类似于一个列表，其中每个元素表示一行数据，由于我们使用了 SELECT *，所以它包含所有数据列。我们在 for 循环中遍历所有列表元素来创建一个包含所有图书信息的字典列表。最后，使用 Flask 的 jsonify() 方法将字典列表转换为 JSON 格式，并在 HTTP 响应中返回它。

现在在浏览器中访问 http://127.0.0.1:5000/books，将会显示包含所有图书信息的 JSON 格式数据。

获取单个图书

让我们现在创建一个路由，它将允许我们获取 book 表中特定 ID 的图书的信息。这也是一个“读取”或“R”操作。

在 app.py 中插入以下代码：

@app.route('/books/<int:book_id>', methods=['GET'])
def get_book(book_id):
    conn = sqlite3.connect('books.db')
    c = conn.cursor()

    c.execute('SELECT * FROM book WHERE id = ?', (book_id,))

    row = c.fetchone()
    if row is None:
        conn.close()
        return jsonify({'error': 'Book not found'}), 404

    book = {
        'id': row[0],
        'title': row[1],
        'author': row[2],
        'isbn': row[3]
    }

    conn.close()

    return jsonify(book)

这个代码定义了 /books/<int:book_id> 路由，它使用 <int:book_id> 路由参数来指定要检索的书籍的 ID。它使用与 get_all_books() 相同的 SQLite3 查询来检索特定的书籍。如果书籍不存在，则返回一个错误消息。否则，这个函数将创建一个包含书籍信息的字典，并将其转换为 JSON 格式并在 HTTP 响应中返回。

现在，在浏览器中访问 http://127.0.0.1:5000/books/1，将会显示 ID 为 1 的书籍的信息。

创建新图书

让我们现在创建一个路由，它将允许我们创建一个新的 book 表记录。这是一个“创建”或“C”操作。

在 app.py 中插入以下代码：

@app.route('/books', methods=['POST'])
def create_book():
    if not request.json:
        return jsonify({'error': 'Data not provided'}), 400

    title = request.json.get('title')
    author = request.json.get('author')
    isbn = request.json.get('isbn')

    if not title or not author or not isbn:
        return jsonify({'error': 'Missing data'}), 400

    conn = sqlite3.connect('books.db')
    c = conn.cursor()

    c.execute('INSERT INTO book (title, author, isbn) VALUES (?, ?, ?)', (title, author, isbn))

    book_id = c.lastrowid

    conn.commit()
    conn.close()

    return jsonify({'id': book_id, 'title': title, 'author': author, 'isbn': isbn}), 201

这个代码定义了 /books 路由，并使用 POST HTTP 方法来创建一个新的 book 表记录。它使用 Flask 的 request 对象来获取请求数据，并检查是否提供了正确的数据。如果数据丢失，则返回一个错误消息。否则，它使用 SQLite3 的 connect() 和 cursor() 方法来连接到数据库并执行 SQL 查询来插入新的记录。它还使用 lastrowid 属性来获取新记录的 ID，并将其返回到响应中。

现在，在 API 客户端（如 Postman）中创建一个新 POST 请求，请求 URL 为 http://localhost:5000/books，并将请求数据发送到该 URL。您应该会在响应中看到您所创建的新书籍的信息。

更新图书

让我们现在创建一个路由，它将允许我们更新 book 表中特定 ID 的图书的信息。这是一个“更新”或“U”操作。

在 app.py 中插入以下代码：

@app.route('/books/<int:book_id>', methods=['PUT'])
def update_book(book_id):
    if not request.json:
        return jsonify({'error': 'Data not provided'}), 400

    title = request.json.get('title')
    author = request.json.get('author')
    isbn = request.json.get('isbn')

    if not title and not author and not isbn:
        return jsonify({'error': 'Missing data'}), 400

    conn = sqlite3.connect('books.db')
    c = conn.cursor()

    c.execute('SELECT * FROM book WHERE id = ?', (book_id,))
    row = c.fetchone()
    if row is None:
        conn.close()
        return jsonify({'error': 'Book not found'}), 404

    fields = []
    if title:
        fields.append(('title', title))
    if author:
        fields.append(('author', author))
    if isbn:
        fields.append(('isbn', isbn))

    query = 'UPDATE book SET'
    values = []
    for field in fields:
        query += ' ' + field[0] + ' = ?,'
        values.append(field[1])
    query = query[:-1] + ' WHERE id = ?'
    values.append(book_id)

    c.execute(query, tuple(values))

    conn.commit()
    conn.close()

    return jsonify({'message': 'Book updated'})

这个代码定义了 /books/<int:book_id> 路由，它使用 <int:book_id> 路由参数来指定要更新的书籍的 ID。它使用 Flask 的 request 对象来获取要更新的数据，并使用 SQLite3 的 connect() 和 cursor() 方法来连接到数据库并查询指定 ID 的书籍的信息。如果找不到书籍，则返回一个错误消息。否则，它将构建一个适当的 SQL 查询，然后执行此查询来更新数据。

现在，在 API 客户端（如 Postman）中创建一个新 PUT 请求，请求 URL 为 http://localhost:5000/books/1（其中 1 是您要更新的书籍的 ID），并将要更新的数据发送到该 URL。您应该会在响应中看到“Book updated”消息。

删除图书

最后，让我们创建一个路由来删除 book 表中特定 ID 的图书的信息。这是一个“删除”或“D”操作。

在 app.py 中插入以下代码：

@app.route('/books/<int:book_id>', methods=['DELETE'])
def delete_book(book_id):
    conn = sqlite3.connect('books.db')
    c = conn.cursor()

    c.execute('SELECT * FROM book WHERE id = ?', (book_id,))
    row = c.fetchone()
    if row is None:
        conn.close()
        return jsonify({'error': 'Book not found'}), 404

    c.execute('DELETE FROM book WHERE id = ?', (book_id,))

    conn.commit()
    conn.close()

    return jsonify({'message': 'Book deleted'})

这个代码定义了一个 /books/<int:book_id> 路由，它使用 <int:book_id> 路由参数来指定要删除的书籍的 ID。它使用 SQLite3 的 connect() 和 cursor() 方法来连接到数据库并查询指定 ID 的书籍的信息。如果找不到书籍，则返回一个错误消息。否则，它将执行一个 DELETE SQL 命令，将找到的记录从表中删除。

现在，在 API 客户端（如 Postman）中创建一个新 DELETE 请求，请求 URL 为 http://localhost:5000/books/1（其中 1 是您要删除的书籍的 ID）。您应该会在响应中看到“Book deleted”消息。

结论

本文提供了一个使用 Python 的 SQLite3 数据库和烧瓶 Web 应用程序进行 CRUD 操作的示例。我们创建了一个包含 book 表的简单数据库，并使用烧瓶应用程序将其公开为 Web API。我们创建了用于“创建”、“读取”、“更新”和“删除” book 表记录的路由，并演示了如何使用 Postman 等工具来测试这些路由。

这样，我们便使用了 Flask 搭建一个从操作到数据库的完整的 Web 应用程序。