文章

Flask - API 版本控制

發布於
作者 Ned Hsu 6 分鐘閱讀

目標

  • 理解 API 版本控制的重要性
  • 使用藍圖實現版本化的端點
  • 設計一個 v2 版本的待辦事項 API

步驟

  1. 準備環境

    • 繼續使用 flask_api/ 項目結構，確保虛擬環境已激活：
      1
2

      # Windows: flask_api_env\Scripts\activate
# macOS/Linux: source flask_api_env/bin/activate

  2. 理解版本控制

    • API 版本控制允許您在不破壞現有客戶端的情況下升級功能。例如，v1 保持穩定，v2 可以引入新功能或更改行為。

  3. 重構現有藍圖

    • 將現有的 todos 和 users 藍圖移動到 v1 版本目錄：
      1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

      flask_api/
├── app/
│   ├── __init__.py
│   ├── routes/
│   │   ├── __init__.py
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── todos.py  # 原 todos.py
│   │   │   └── users.py  # 原 users.py
│   │   └── v2/
│   │       ├── __init__.py
│   │       └── todos.py  # 新版本
│   ├── models.py
│   ├── schemas.py
│   └── config.py
└── run.py

    • 更新 app/init.py，註冊 v1 和 v2 藍圖：

      1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

      from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
from .routes.v1.todos import todos_bp as todos_v1_bp
from .routes.v1.users import users_bp as users_v1_bp
from .routes.v2.todos import todos_bp as todos_v2_bp  # 新增 v2

db = SQLAlchemy()
ma = Marshmallow()

def create_app():
    app = Flask(__name__)
    app.config.from_object('app.config.Config')
    db.init_app(app)
    ma.init_app(app)

    # 註冊藍圖
    app.register_blueprint(todos_v1_bp, url_prefix='/api/v1')
    app.register_blueprint(users_v1_bp, url_prefix='/api/v1')
    app.register_blueprint(todos_v2_bp, url_prefix='/api/v2')  # 新增 v2

    @app.errorhandler(404)
    def not_found(error):
        return jsonify({'error': 'Not Found', 'message': str(error)}), 404

    @app.errorhandler(400)
    def bad_request(error):
        return jsonify({'error': 'Bad Request', 'message': str(error)}), 400

    @app.errorhandler(500)
    def internal_error(error):
        return jsonify({'error': 'Internal Server Error', 'message': 'Something went wrong on our end'}), 500

    with app.app_context():
        db.create_all()

    return app

  4. 創建 v2 版本的待辦事項 API

    • 新建 app/routes/v2/todos.py，改進 v2 版本（例如添加任務優先級）：

      1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

      from flask import Blueprint, jsonify, request, abort
from ...models import Todo, User
from ... import db
from ...schemas import todo_schema, todos_schema

todos_bp = Blueprint('todos_v2', __name__)

@todos_bp.route('/todos', methods=['GET'])
def get_todos():
    completed = request.args.get('completed', type=lambda x: x.lower() == 'true')
    user_id = request.args.get('user_id', type=int)
    priority = request.args.get('priority', type=int)  # 新增優先級過濾
    query = Todo.query
    if completed is not None:
        query = query.filter_by(completed=completed)
    if user_id:
        query = query.filter_by(user_id=user_id)
    if priority is not None:
        query = query.filter_by(priority=priority)
    todos = query.all()
    return jsonify({'todos': todos_schema.dump(todos)})

@todos_bp.route('/todos/<int:todo_id>', methods=['GET'])
def get_todo(todo_id):
    todo = Todo.query.get_or_404(todo_id, description='Todo not found')
    return jsonify(todo_schema.dump(todo))

@todos_bp.route('/todos', methods=['POST'])
def create_todo():
    if not request.is_json:
        abort(400, description='Request must be JSON')
    data = request.get_json()
    if 'title' not in data or 'user_id' not in data:
        abort(400, description='Missing title or user_id')
    if not User.query.get(data['user_id']):
        abort(400, description='User not found')
    todo = Todo(
        title=data['title'],
        completed=data.get('completed', False),
        description=data.get('description'),
        user_id=data['user_id'],
        priority=data.get('priority', 0)  # 新增優先級，默認為 0
    )
    db.session.add(todo)
    db.session.commit()
    return jsonify(todo_schema.dump(todo)), 201

@todos_bp.route('/todos/<int:todo_id>', methods=['PUT'])
def update_todo(todo_id):
    todo = Todo.query.get_or_404(todo_id, description='Todo not found')
    if not request.is_json:
        abort(400, description='Request must be JSON')
    data = request.get_json()
    if 'title' in data:
        todo.title = data['title']
    if 'completed' in data and isinstance(data['completed'], bool):
        todo.completed = data['completed']
    if 'description' in data:
        todo.description = data['description']
    if 'user_id' in data:
        if not User.query.get(data['user_id']):
            abort(400, description='User not found')
        todo.user_id = data['user_id']
    if 'priority' in data and isinstance(data['priority'], int):
        todo.priority = data['priority']
    db.session.commit()
    return jsonify(todo_schema.dump(todo)), 200

@todos_bp.route('/todos/<int:todo_id>', methods=['DELETE'])
def delete_todo(todo_id):
    todo = Todo.query.get_or_404(todo_id, description='Todo not found')
    db.session.delete(todo)
    db.session.commit()
    return jsonify({'message': 'Todo deleted'}), 200

    • 注意：這裡假設 Todo 模型已添加 priority 字段，需更新模型。

  5. 更新模型和序列化器

    • 修改 app/models.py，添加 priority 字段：

      1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

      from . import db
from datetime import datetime

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(50), unique=True, nullable=False)
    todos = db.relationship('Todo', backref='user', lazy=True)

    def to_dict(self):
        return {'id': self.id, 'username': self.username}

class Todo(db.Model):
    __tablename__ = 'todos'
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    completed = db.Column(db.Boolean, default=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    description = db.Column(db.String(255), nullable=True)
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
    priority = db.Column(db.Integer, default=0)  # 新增優先級字段

    def to_dict(self):
        return {
            'id': self.id,
            'title': self.title,
            'completed': self.completed,
            'created_at': self.created_at.isoformat(),
            'description': self.description,
            'user_id': self.user_id,
            'priority': self.priority
        }

    • 更新 app/schemas.py，包含 priority：

      1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

      from . import ma
from .models import User, Todo

class TodoSchema(ma.SQLAlchemyAutoSchema):
    class Meta:
        model = Todo
        include_fk = True
        fields = ('id', 'title', 'completed', 'created_at', 'description', 'user_id', 'priority')

class UserSchema(ma.SQLAlchemyAutoSchema):
    class Meta:
        model = User
        fields = ('id', 'username', 'todos')
    todos = ma.Nested('TodoSchema', many=True)

todo_schema = TodoSchema()
todos_schema = TodoSchema(many=True)
user_schema = UserSchema()
users_schema = UserSchema(many=True)

  6. 運行應用

    • 刪除舊的 todos.db（因表結構改變），然後運行：
      1

      python run.py

  7. 測試 API

    • 使用 Postman 測試：
      • POST /api/v1/users：
        • Body：{"username": "alice"}
      • POST /api/v2/todos：
        • Body：{"title": "Learn Flask", "user_id": 1, "priority": 2}
        • 預期響應：{"id": 1, "title": "Learn Flask", ..., "priority": 2}
      • GET /api/v2/todos?priority=2：
        • 預期響應：僅包含優先級為 2 的任務。
      • GET /api/v1/todos：
        • 確認 v1 版本仍正常工作（不含 priority 過濾）。

  8. 作業

    • 在 v2 中添加一個新端點 GET /api/v2/todos/priority，返回按優先級降序排序的任務。
    • 比較 v1 和 v2 的 GET /todos 響應，確保版本間差異清晰。

注意事項

  • 表結構改變後需重建數據庫。
  • 確保藍圖名稱唯一（例如 todos_v2），避免衝突。
Software, Python
Flask
本文章以 CC BY 4.0 授權