文章

Flask - 數據序列化

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

目標

  • 安裝並配置 Flask-Marshmallow
  • 定義序列化模式（Schema）
  • 在 API 中使用序列化器處理數據

步驟

  1. 準備環境

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

      # Windows: flask_api_env\Scripts\activate
# macOS/Linux: source flask_api_env/bin/activate
    • 安裝 Flask-Marshmallow：
      1

      pip install flask-marshmallow

  2. 配置 Flask-Marshmallow

    • 修改 app/init.py，初始化 Marshmallow：

      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

      from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow  # 新增
from .routes.todos import todos_bp
from .routes.users import users_bp

db = SQLAlchemy()
ma = Marshmallow()  # 初始化 Marshmallow

def create_app():
    app = Flask(__name__)
    app.config.from_object('app.config.Config')
    db.init_app(app)
    ma.init_app(app)  # 初始化 Marshmallow

    app.register_blueprint(todos_bp, url_prefix='/api/v1')
    app.register_blueprint(users_bp, url_prefix='/api/v1')

    @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

  3. 定義序列化模式

    • 新建 app/schemas.py，定義 User 和 Todo 的序列化器：

      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  # 包含外鍵 (user_id)
        fields = ('id', 'title', 'completed', 'created_at', 'description', 'user_id')

class UserSchema(ma.SQLAlchemyAutoSchema):
    class Meta:
        model = User
        fields = ('id', 'username', 'todos')
    todos = ma.Nested('TodoSchema', many=True)  # 嵌套 Todo 數據

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

    • 代碼解釋：

      • ma.SQLAlchemyAutoSchema：自動從模型生成序列化規則。
      • ma.Nested：嵌套關聯數據（例如用戶的任務）。
      • many=True：處理多個對象的序列化。

  4. 更新路由

    • 修改 app/routes/todos.py，使用序列化器：

      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

      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', __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)
    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)
    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']
    )
    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']
    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

    • 修改 app/routes/users.py：

      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

      from flask import Blueprint, jsonify, request, abort
from ..models import User
from .. import db
from ..schemas import user_schema, users_schema

users_bp = Blueprint('users', __name__)

@users_bp.route('/users', methods=['GET'])
def get_users():
    users = User.query.all()
    return jsonify({'users': users_schema.dump(users)})

@users_bp.route('/users', methods=['POST'])
def create_user():
    if not request.is_json:
        abort(400, description='Request must be JSON')
    data = request.get_json()
    if 'username' not in data:
        abort(400, description='Missing username')
    if User.query.filter_by(username=data['username']).first():
        abort(400, description='Username already exists')
    user = User(username=data['username'])
    db.session.add(user)
    db.session.commit()
    return jsonify(user_schema.dump(user)), 201

  5. 運行應用

    • 運行：
      1

      python run.py

  6. 測試 API

    • 使用 Postman 測試：
      • POST /api/v1/users：
        • Body：{"username": "alice"}
        • 預期響應：{"id": 1, "username": "alice", "todos": []}
      • POST /api/v1/todos：
        • Body：{"title": "Learn Flask", "user_id": 1, "description": "Study Flask"}
        • 預期響應：{"id": 1, "title": "Learn Flask", "completed": false, "created_at": "...", "description": "Study Flask", "user_id": 1}
      • GET /api/v1/users：
        • 預期響應：{"users": [{"id": 1, "username": "alice", "todos": [{"id": 1, "title": "Learn Flask", ...}]}]}
      • GET /api/v1/todos：
        • 預期響應：{"todos": [{"id": 1, "title": "Learn Flask", ...}]}

  7. 作業

    • 在 TodoSchema 中添加一個自定義字段，例如 user_username，顯示關聯用戶的用戶名（提示：使用 @post_dump）。
    • 添加一個端點 GET /api/v1/users/<int:user_id>，返回單個用戶及其任務。

注意事項

  • dump() 方法將模型轉換為 Python 字典，需用 jsonify 包裝以返回 JSON。
  • 如果遇到序列化錯誤，檢查模型字段是否與 Schema 一致。
Software, Python
Flask
本文章以 CC BY 4.0 授權