文章

Flask - SQLAlchemy 模型關係

目標

  • 理解一對多關係
  • 在模型中配置外鍵
  • 查詢和管理關聯數據

步驟

  1. 準備環境

    • 繼續使用 flask_api/ 項目結構,確保虛擬環境已激活:
      1
      2
      
      # Windows: flask_api_env\Scripts\activate
      # macOS/Linux: source flask_api_env/bin/activate
      
  2. 更新模型

    • 修改 app/models.py,添加 User 模型並建立與 Todo 的關係:

      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
      
      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)  # 外鍵
      
          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
              }
      
    • 代碼解釋

      • db.relationship:定義 User 和 Todo 之間的一對多關係,backref 允許從 Todo 反向訪問 User。
      • db.ForeignKey:在 Todo 中添加外鍵,指向 User 的 id
      • 由於表結構改變,需刪除現有的 todos.db 文件並重建。
  3. 更新應用初始化

    • app/init.py 已包含 db.create_all(),無需修改。
  4. 添加用戶路由

    • 新建 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
      26
      
      from flask import Blueprint, jsonify, request, abort
      from ..models import User
      from .. import db
      
      users_bp = Blueprint('users', __name__)
      
      # GET - 獲取所有用戶
      @users_bp.route('/users', methods=['GET'])
      def get_users():
          users = User.query.all()
          return jsonify({'users': [user.to_dict() for user in users]})
      
      # POST - 創建新用戶
      @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.to_dict()), 201
      
    • app/init.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
      
      from flask import Flask, jsonify
      from flask_sqlalchemy import SQLAlchemy
      from .routes.todos import todos_bp
      from .routes.users import users_bp  # 新增
      
      db = SQLAlchemy()
      
      def create_app():
          app = Flask(__name__)
          app.config.from_object('app.config.Config')
          db.init_app(app)
      
          # 註冊藍圖
          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
      
  5. 更新待辦事項路由

    • 修改 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
      69
      70
      71
      72
      
      from flask import Blueprint, jsonify, request, abort
      from ..models import Todo, User
      from .. import db
      
      todos_bp = Blueprint('todos', __name__)
      
      # GET - 獲取所有任務
      @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': [todo.to_dict() for todo in todos]})
      
      # GET - 獲取單個任務
      @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.to_dict())
      
      # POST - 創建新任務
      @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.to_dict()), 201
      
      # PUT - 更新任務
      @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.to_dict()), 200
      
      # DELETE - 刪除任務
      @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
      
  6. 運行應用

    • 刪除舊的 todos.db(如果存在),然後運行:
      1
      
      python run.py
      
  7. 測試 API

    • 使用 Postman 測試:
      • POST /api/v1/users
        • Body:{"username": "alice"}
        • 預期響應:{"id": 1, "username": "alice"}
      • POST /api/v1/users
        • Body:{"username": "bob"}
      • POST /api/v1/todos
        • Body:{"title": "Learn Flask", "user_id": 1, "description": "Study Flask basics"}
        • 預期響應:{"id": 1, "title": "Learn Flask", "completed": false, "created_at": "...", "description": "Study Flask basics", "user_id": 1}
      • GET /api/v1/todos?user_id=1
        • 預期響應:僅包含 user_id 為 1 的任務。
      • PUT /api/v1/todos/1
        • Body:{"user_id": 2}
        • 預期響應:任務的 user_id 更新為 2。
      • GET /api/v1/users
        • 預期響應:{"users": [{"id": 1, "username": "alice"}, {"id": 2, "username": "bob"}]}
  8. 作業

    • 修改 User.to_dict(),返回該用戶的所有任務(提示:遍歷 self.todos)。
    • 添加一個端點 GET /api/v1/users/<int:user_id>/todos,返回指定用戶的所有任務。

注意事項

  • 表結構改變後需重建數據庫。
  • 外鍵約束確保 user_id 必須對應存在的用戶。

本文章以 CC BY 4.0 授權