RESTful API — 让你的应用会说话
🕐 预计用时:2-3 小时 | 🎯 目标:掌握 Blueprint 模块化、JSON API 设计规范和状态码
📖 今日目录
1. 什么是 RESTful API?
🌐 一句话解释
RESTful API 是前后端之间的"合同"——前端(网页/手机)用 HTTP 请求向后端要数据,后端返回 JSON 格式的数据。
# 前端:"给我所有用户"
GET /api/users
→ [{"id":1,"name":"alice"}, {"id":2,"name":"bob"}]
# 前端:"给我 id=1 的用户"
GET /api/users/1
→ {"id":1,"name":"alice","email":"alice@example.com"}
# 前端:"创建一个新用户"
POST /api/users body: {"name":"charlie","email":"c@example.com"}
→ {"id":3,"name":"charlie"} (201 Created)
# 前端:"删除 id=1 的用户"
DELETE /api/users/1
→ 204 No Content
🤝 REST 的核心原则
| | |
|---|
| 资源导向 | | /api/users |
| HTTP 动词 | | GET 读 / POST 创建 / PUT 更新 / DELETE 删除 |
| 无状态 | | |
| JSON 格式 | | {"name":"alice"} |
📊 URL 设计对照表
| | | |
|---|
| | /api/users | |
| | /api/users/1 | |
| | /api/users | |
| | /api/users/1 | |
| | /api/users/1 | |
| | /api/users?q=alice | |
2. HTTP 状态码
状态码是后端给前端的"回执"——告诉前端请求的结果。
📋 常用状态码速查
| | |
|---|
| 200 | | |
| 201 | | |
| 204 | | |
| 400 | | |
| 401 | | |
| 403 | | |
| 404 | | |
| 409 | | |
| 422 | | |
| 500 | | |
💡 状态码的记忆口诀:
• 2xx:成功(200 OK / 201 Created / 204 No Content)
• 4xx:客户端的错(400 参数错 / 401 没登录 / 403 没权限 / 404 不存在)
• 5xx:服务器的错(500 代码 bug / 502 网关错 / 503 服务不可用)
3. Blueprint 蓝图
🧩 为什么需要蓝图?
当应用变大时,所有路由写在一个文件里会变成几千行的"上帝文件"。蓝图让你把路由按功能拆分。
# 项目结构
myapp/
├── app.py
├── api/
│ ├── __init__.py
│ ├── users.py ← 用户相关 API
│ └── posts.py ← 文章相关 API
├── models.py
└── templates/
🔧 创建蓝图
# api/users.py
from flask import Blueprint, jsonify, request
# 创建蓝图:名称 'users',URL 前缀 /api/users
users_bp = Blueprint('users', __name__, url_prefix='/api/users')
@users_bp.route('/', methods=['GET'])
def get_users():
return jsonify([{'id': 1, 'name': 'alice'}])
@users_bp.route('/<int:user_id>', methods=['GET'])
def get_user(user_id):
return jsonify({'id': user_id, 'name': 'alice'})
@users_bp.route('/', methods=['POST'])
def create_user():
data = request.get_json()
return jsonify({'id': 3, 'name': data['name']}), 201
📌 注册蓝图
# app.py
from flask import Flask
from api.users import users_bp
from api.posts import posts_bp
app = Flask(__name__)
# 注册蓝图
app.register_blueprint(users_bp)
app.register_blueprint(posts_bp)
# 最终 URL:
# GET /api/users/ → users_bp.get_users()
# GET /api/users/1 → users_bp.get_user(1)
# GET /api/posts/ → posts_bp.get_posts()
💡 蓝图 = 路由的"文件夹"
把相关路由放在同一个蓝图里,就像把文件分类放到不同文件夹。一个蓝图一个文件,改某个功能只动一个文件。
类比:蓝图 = 部门(销售部、技术部、财务部),每个部门有自己的职责和办公区域。
4. 第一个 API
🎯 最小 RESTful API
# app.py
from flask import Flask, jsonify, request
app = Flask(__name__)
# 模拟数据库
users = [
{'id': 1, 'name': 'alice', 'email': 'alice@example.com'},
{'id': 2, 'name': 'bob', 'email': 'bob@example.com'},
]
# GET /api/users → 获取所有用户
@app.route('/api/users', methods=['GET'])
def get_users():
return jsonify({
'code': 200,
'data': users,
'total': len(users)
})
# GET /api/users/<id> → 获取单个用户
@app.route('/api/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
user = next((u for u in users if u['id'] == user_id), None)
if not user:
return jsonify({'code': 404, 'message': '用户不存在'}), 404
return jsonify({'code': 200, 'data': user})
# POST /api/users → 创建用户
@app.route('/api/users', methods=['POST'])
def create_user():
data = request.get_json()
if not data or 'name' not in data:
return jsonify({'code': 400, 'message': '缺少 name 字段'}), 400
new_user = {
'id': len(users) + 1,
'name': data['name'],
'email': data.get('email', '')
}
users.append(new_user)
return jsonify({'code': 201, 'data': new_user}), 201
if __name__ == '__main__':
app.run(debug=True)
🧪 测试 API(用 curl)
# 获取所有用户
curl http://127.0.0.1:5000/api/users
# {"code":200,"data":[{"id":1,"name":"alice"},...],"total":2}
# 获取单个用户
curl http://127.0.0.1:5000/api/users/1
# {"code":200,"data":{"id":1,"name":"alice","email":"alice@example.com"}}
# 创建用户
curl -X POST http://127.0.0.1:5000/api/users \
-H "Content-Type: application/json" \
-d '{"name":"charlie","email":"c@example.com"}'
# {"code":201,"data":{"id":3,"name":"charlie","email":"c@example.com"}}
5. 完整 CRUD API
# api/users.py — 完整的用户 CRUD API
from flask import Blueprint, jsonify, request
from models import db, User
users_bp = Blueprint('users', __name__, url_prefix='/api/users')
# GET /api/users — 获取用户列表(支持分页和搜索)
@users_bp.route('/', methods=['GET'])
def get_users():
page = request.args.get('page', 1, type=int)
per_page = request.args.get('per_page', 10, type=int)
keyword = request.args.get('q', '')
query = User.query
if keyword:
query = query.filter(User.username.like(f'%{keyword}%'))
pagination = query.paginate(page=page, per_page=per_page)
return jsonify({
'code': 200,
'data': [{'id': u.id, 'name': u.username, 'email': u.email} for u in pagination.items],
'total': pagination.total,
'pages': pagination.pages,
'current_page': page
})
# GET /api/users/<id> — 获取单个用户
@users_bp.route('/<int:user_id>', methods=['GET'])
def get_user(user_id):
user = User.query.get_or_404(user_id)
return jsonify({
'code': 200,
'data': {'id': user.id, 'name': user.username, 'email': user.email}
})
# POST /api/users — 创建用户
@users_bp.route('/', methods=['POST'])
def create_user():
data = request.get_json()
if not data:
return jsonify({'code': 400, 'message': '请求体为空'}), 400
if not data.get('username') or not data.get('email'):
return jsonify({'code': 400, 'message': 'username 和 email 必填'}), 400
if User.query.filter_by(username=data['username']).first():
return jsonify({'code': 409, 'message': '用户名已存在'}), 409
user = User(username=data['username'], email=data['email'])
if data.get('password'):
user.set_password(data['password'])
db.session.add(user)
db.session.commit()
return jsonify({'code': 201, 'data': {'id': user.id, 'name': user.username}}), 201
# PUT /api/users/<id> — 更新用户
@users_bp.route('/<int:user_id>', methods=['PUT'])
def update_user(user_id):
user = User.query.get_or_404(user_id)
data = request.get_json()
if data.get('username'):
user.username = data['username']
if data.get('email'):
user.email = data['email']
db.session.commit()
return jsonify({'code': 200, 'data': {'id': user.id, 'name': user.username}})
# DELETE /api/users/<id> — 删除用户
@users_bp.route('/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.commit()
return '', 204
6. API 设计规范
📐 统一响应格式
# 推荐的统一响应格式
{
"code": 200, // 业务状态码
"message": "success", // 提示信息
"data": { ... }, // 数据(成功时)
"errors": [...] // 错误详情(失败时)
}
# 成功示例
{"code": 200, "message": "success", "data": {"id": 1, "name": "alice"}}
# 错误示例
{"code": 400, "message": "参数错误", "errors": ["username 不能为空", "email 格式不对"]}
📋 API 设计 Checklist
| | |
|---|
| /api/users | /api/getUser |
| /api/users/1/posts | /api/getUserPosts?uid=1 |
| /api/v1/users | /api/users?version=1 |
| | |
| | 200 + {"status":"created"} |
| | |
7. 错误处理
# 全局错误处理器
@app.errorhandler(404)
def not_found(error):
return jsonify({'code': 404, 'message': '资源不存在'}), 404
@app.errorhandler(500)
def internal_error(error):
return jsonify({'code': 500, 'message': '服务器内部错误'}), 500
@app.errorhandler(400)
def bad_request(error):
return jsonify({'code': 400, 'message': '请求参数有误'}), 400
# abort 主动抛出错误
from flask import abort
@users_bp.route('/<int:user_id>')
def get_user(user_id):
user = User.query.get(user_id)
if not user:
abort(404) # 自动触发 404 处理器
return jsonify(user.to_dict())
8. 分页
# 分页查询
@users_bp.route('/', methods=['GET'])
def get_users():
page = request.args.get('page', 1, type=int)
per_page = request.args.get('per_page', 20, type=int)
# 限制每页数量
per_page = min(per_page, 100)
pagination = User.query.paginate(page=page, per_page=per_page)
return jsonify({
'data': [u.to_dict() for u in pagination.items],
'pagination': {
'page': pagination.page,
'per_page': pagination.per_page,
'total': pagination.total,
'pages': pagination.pages,
'has_next': pagination.has_next,
'has_prev': pagination.has_prev,
}
})
9. 今日练习
🏋️ 练习 1:文章 CRUD API
# 实现文章的完整 CRUD:
# GET /api/posts → 获取文章列表(分页)
# GET /api/posts/<id> → 获取单篇文章
# POST /api/posts → 创建文章
# PUT /api/posts/<id> → 更新文章
# DELETE /api/posts/<id> → 删除文章
# 要求:统一响应格式 + 合适的状态码
🏋️ 练习 2:嵌套资源
# 实现用户-文章的嵌套 API:
# GET /api/users/<id>/posts → 获取某用户的所有文章
# POST /api/users/<id>/posts → 给某用户创建文章
# 要求:检查用户是否存在(404)
🏋️ 练习 3:搜索与过滤
# 给文章 API 添加搜索和过滤:
# GET /api/posts?q=keyword → 关键词搜索
# GET /api/posts?sort=created_at → 按时间排序
# GET /api/posts?sort=views → 按浏览量排序
# GET /api/posts?order=desc → 降序
10. 今日小结
| |
|---|
| 资源导向 + HTTP 动词 + JSON 格式 + 无状态 |
| 2xx 成功 / 4xx 客户端错 / 5xx 服务器错 |
| 模块化路由,一个蓝图一个文件,register_blueprint 注册 |
| GET 查询 / POST 创建 / PUT 更新 / DELETE 删除 |
| { code, message, data } 统一格式 |
| errorhandler + abort + jsonify |
| paginate(page, per_page) + 分页元数据 |
🚀 明日预告:Day 57 — Flask 进阶
API 写好了,但还需要更多"工程化"能力:中间件、错误处理、日志系统、CORS 跨域、配置管理。让你的 Flask 应用从"能跑"变成"能上线"!