路由设置

修改 main.py，为 Book 添加 CRUD 路由：

添加 CRUD 操作

from datetime import datetime
import sqlite3

from flask import Flask, request, g, jsonify

app = Flask(__name__)
DATABASE = 'test.db'

def get_db():
    db = getattr(g, '_database', None)
    if db is None:
        db = sqlite3.connect(DATABASE)
        g._database = db
        cursor = db.cursor()
        cursor.execute('''CREATE TABLE IF NOT EXISTS books (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            author TEXT NOT NULL,
            published_at TEXT NOT NULL,
            description TEXT NOT NULL,
            isbn TEXT NOT NULL,
            total_pages INTEGER NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
          )''')
        cursor.close()
    return db

@app.teardown_appcontext
def close_connection(exception):
    db = getattr(g, '_database', None)
    if db is not None:
        db.close()

def execute_query(query, values=()):
    cursor = get_db().cursor()
    cursor.execute(query, values)
    result = cursor.fetchall()
    get_db().commit()
    cursor.close()
    return result

# Define a health endpoint handler, use `/health` or `/`
@app.route('/')
def health():
    # Return a simple response indicating the server is healthy
    return {"status": "ok"}

@app.route('/books', methods=['GET'])
def get_books():
    query = "SELECT * FROM books"
    books = execute_query(query)
    return books

@app.route('/books/<int:book_id>', methods=['GET'])
def get_book(book_id):
    query = "SELECT * FROM books WHERE id = ?"
    books = execute_query(query, (book_id,))
    if not books:
        return {"error": "Record not found"}, 404
    return jsonify(books[0])

@app.route('/books', methods=['POST'])
def create_book():
    b = request.get_json()
    query = "INSERT INTO books (title, author, published_at, description, isbn, total_pages, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
    now = datetime.now()
    values = (b['title'], b['author'], b['published_at'], b['description'], b['isbn'], b['total_pages'], now, now)
    execute_query(query, values)
    return b, 201

@app.route('/books/<int:book_id>', methods=['PUT'])
def update_book(book_id):
    b = request.get_json()
    query = "UPDATE books SET title=?, author=?, published_at=?, description=?, isbn=?, total_pages=?, updated_at=? WHERE id=?"
    values = (b['title'], b['author'], b['published_at'], b['description'], b['isbn'], b['total_pages'], datetime.now(), book_id)
    execute_query(query, values)
    return b

@app.route('/books/<int:book_id>', methods=['DELETE'])
def delete_book(book_id):
    query = "DELETE FROM books WHERE id=?"
    execute_query(query, (book_id,))
    return '', 204

此刻，暂使用 SQLite¹ 数据库做演示。

sqlite3 是 Python 自带的数据库模块，用于与 SQLite 数据库交互。SQLite 是一种轻量级、无服务器、自包含的 SQL 数据库引擎。

curl 测试

创建一本新图书：

curl -X POST \
  http://localhost:5000/books \
  -H 'Content-Type: application/json' \
  -d '{
    "title": "Sample Book",
    "author": "John Doe",
    "published_at": "2023-01-01",
    "description": "A sample book description",
    "isbn": "1234567890",
    "total_pages": 200
}'

应如下响应：

{
  "author": "John Doe",
  "description": "A sample book description",
  "isbn": "1234567890",
  "published_at": "2023-01-01",
  "title": "Sample Book",
  "total_pages": 200
}

根据 ID 获取一本图书：

curl -X GET http://localhost:5000/books/1

结果：

[
  1,
  "Sample Book",
  "John Doe",
  "2023-01-01",
  "A sample book description",
  "1234567890",
  200,
  "2024-03-05 21:20:39.938647",
  "2024-03-05 21:20:39.938647"
]

列出所有图书：

curl -X GET http://localhost:5000/books

结果列表：

[
  [
    1,
    "Sample Book",
    "John Doe",
    "2023-01-01",
    "A sample book description",
    "1234567890",
    200,
    "2024-03-05 21:20:39.938647",
    "2024-03-05 21:20:39.938647"
  ],
  [
    2,
    "Great Book",
    "Bob Smith",
    "2003-01-01",
    "A sample book description",
    "1234567890",
    200,
    "2024-03-05 21:29:11.470872",
    "2024-03-05 21:29:11.470872"
  ]
]

更新一本已有的图书：

curl -X PUT \
  http://localhost:5000/books/1 \
  -H 'Content-Type: application/json' \
  -d '{
    "title": "Updated Book Title",
    "author": "Jane Smith",
    "published_at": "2023-01-01",
    "description": "A new description",
    "isbn": "1234567890",
    "total_pages": 200
}'

结果：

{
  "author": "Jane Smith",
  "description": "A new description",
  "isbn": "1234567890",
  "published_at": "2023-01-01",
  "title": "Updated Book Title",
  "total_pages": 200
}

删除一本已有的图书：

curl -X DELETE http://localhost:5000/books/1

服务端返回 code 204 以表示成功删除。

此刻 REST API 服务器已经初具雏形。不错！

Footnotes

1. SQLite: https://www.sqlite.org/index.html ↩