文档
Flask RESTful API —— 构建完整的 CRUD 接口
目标
使用 Flask 构建一个完整的 RESTful API,实现对"书籍"资源的 CRUD 操作:
- 理解 RESTful URL 设计规范
- 掌握请求体 JSON 解析
- 掌握状态码的正确使用
- 使用 Flask-RESTful 扩展简化 API 开发
完整代码
# app_rest.py
from flask import Flask, request, jsonify, abort
from flask_restful import Api, Resource, reqparse, fields, marshal_with
from datetime import datetime
app = Flask(__name__)
api = Api(app)
# ---------- 模拟数据库 ----------
books_db = [
{"id": 1, "title": "1984", "author": "George Orwell", "year": 1949, "created_at": "2024-01-01T00:00:00"},
{"id": 2, "title": "三体", "author": "刘慈欣", "year": 2008, "created_at": "2024-01-02T00:00:00"},
]
# ---------- 请求解析器(输入验证)----------
book_parser = reqparse.RequestParser()
book_parser.add_argument("title", type=str, required=True, help="书名不能为空")
book_parser.add_argument("author", type=str, required=True, help="作者不能为空")
book_parser.add_argument("year", type=int, required=True, help="出版年份必须为整数")
# ---------- 响应格式化 ----------
book_fields = {
"id": fields.Integer,
"title": fields.String,
"author": fields.String,
"year": fields.Integer,
"created_at": fields.String,
}
class BookListResource(Resource):
"""处理 /api/books 的 GET(列表)和 POST(创建)"""
def get(self):
"""获取所有书籍"""
return {"books": books_db, "total": len(books_db)}
@marshal_with(book_fields)
def post(self):
"""创建新书籍"""
args = book_parser.parse_args()
new_id = max(b["id"] for b in books_db) + 1 if books_db else 1
new_book = {
"id": new_id,
"title": args["title"],
"author": args["author"],
"year": args["year"],
"created_at": datetime.now().isoformat(),
}
books_db.append(new_book)
return new_book, 201
class BookResource(Resource):
"""处理 /api/books/<book_id> 的 GET / PUT / DELETE"""
@marshal_with(book_fields)
def get(self, book_id):
"""获取单本书"""
book = next((b for b in books_db if b["id"] == book_id), None)
if book is None:
abort(404, message=f"书籍 {book_id} 不存在")
return book
@marshal_with(book_fields)
def put(self, book_id):
"""更新书籍"""
book = next((b for b in books_db if b["id"] == book_id), None)
if book is None:
abort(404, message=f"书籍 {book_id} 不存在")
args = book_parser.parse_args()
book["title"] = args["title"]
book["author"] = args["author"]
book["year"] = args["year"]
return book
def delete(self, book_id):
"""删除书籍"""
global books_db
book = next((b for b in books_db if b["id"] == book_id), None)
if book is None:
abort(404, message=f"书籍 {book_id} 不存在")
books_db = [b for b in books_db if b["id"] != book_id]
return {"message": f"书籍 {book_id} 已删除"}, 200
# ---------- 注册路由 ----------
api.add_resource(BookListResource, "/api/books")
api.add_resource(BookResource, "/api/books/<int:book_id>")
# ---------- 错误处理 ----------
@app.errorhandler(404)
def handle_404(error):
return jsonify({"error": str(error)}), 404
@app.errorhandler(400)
def handle_400(error):
return jsonify({"error": str(error)}), 400
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000, debug=True)
测试 API
# 获取所有书籍
curl http://localhost:5000/api/books
# 获取单本书
curl http://localhost:5000/api/books/1
# 创建新书
curl -X POST http://localhost:5000/api/books \
-H "Content-Type: application/json" \
-d '{"title":"Animal Farm","author":"George Orwell","year":1945}'
# 更新书籍
curl -X PUT http://localhost:5000/api/books/1 \
-H "Content-Type: application/json" \
-d '{"title":"Nineteen Eighty-Four","author":"George Orwell","year":1949}'
# 删除书籍
curl -X DELETE http://localhost:5000/api/books/2
RESTful URL 设计规范
| 方法 | URL | 操作 | 状态码 |
|---|---|---|---|
| GET | /api/books |
获取列表 | 200 |
| GET | /api/books/1 |
获取单个 | 200 / 404 |
| POST | /api/books |
创建 | 201 |
| PUT | /api/books/1 |
更新 | 200 / 404 |
| DELETE | /api/books/1 |
删除 | 200 / 404 |
关键要点
- reqparse:用于请求参数验证,自动返回 400 错误
- marshal_with:装饰器自动将返回 dict 按 book_fields 格式化
- abort():中断请求并返回错误响应
- 类视图 + add_resource:比装饰器更符合 RESTful 组织方式