Flask - SQLAlchemy 模型關係
目標
- 理解一對多關係
- 在模型中配置外鍵
- 查詢和管理關聯數據
步驟
準備環境
- 繼續使用
flask_api/
項目結構,確保虛擬環境已激活:1 2
# Windows: flask_api_env\Scripts\activate # macOS/Linux: source flask_api_env/bin/activate
- 繼續使用
更新模型
修改 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
文件並重建。
更新應用初始化
- app/init.py 已包含
db.create_all()
,無需修改。
- app/init.py 已包含
添加用戶路由
新建 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
更新待辦事項路由
修改 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
運行應用
- 刪除舊的
todos.db
(如果存在),然後運行:1
python run.py
- 刪除舊的
測試 API
- 使用 Postman 測試:
- POST /api/v1/users:
- Body:
{"username": "alice"}
- 預期響應:
{"id": 1, "username": "alice"}
- Body:
- POST /api/v1/users:
- Body:
{"username": "bob"}
- Body:
- 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}
- Body:
- GET /api/v1/todos?user_id=1:
- 預期響應:僅包含 user_id 為 1 的任務。
- PUT /api/v1/todos/1:
- Body:
{"user_id": 2}
- 預期響應:任務的
user_id
更新為 2。
- Body:
- GET /api/v1/users:
- 預期響應:
{"users": [{"id": 1, "username": "alice"}, {"id": 2, "username": "bob"}]}
- 預期響應:
- POST /api/v1/users:
- 使用 Postman 測試:
作業
- 修改
User.to_dict()
,返回該用戶的所有任務(提示:遍歷self.todos
)。 - 添加一個端點
GET /api/v1/users/<int:user_id>/todos
,返回指定用戶的所有任務。
- 修改
注意事項
- 表結構改變後需重建數據庫。
- 外鍵約束確保
user_id
必須對應存在的用戶。
本文章以 CC BY 4.0 授權