iT邦幫忙

2026 iThome 鐵人賽
DAY 29
0
Software Development

從零開始的後端開發：用 Flask 實作 REST API，搞懂 API 與資料庫之間如何協作系列 第 29 篇

Day 29｜把全部串起來

18th鐵人賽
53 瀏覽

  • 分享至 

  • xImage
    •  

前面 28 天，我們是一塊一塊慢慢加上去的：

先寫 Flask，再換成 Flask-RESTX，接著接資料庫、加 Migration，最後加上錯誤處理和資安。

每天都只看一小塊，今天就把它們全部放在一起，看看完整的樣子。

這個 API 現在有什麼？

功能：

GET /tasks 取得所有作業（可以用 completed 篩選）
POST /tasks 新增作業（需要金鑰）
GET /tasks/{id} 取得單筆作業
PUT /tasks/{id} 修改作業（需要金鑰）
DELETE /tasks/{id} 刪除作業（需要管理員金鑰）

用到的東西：

Flask + Flask-RESTX → API 和 Swagger 文件
Flask-SQLAlchemy → 資料庫
Flask-Migrate → 資料表結構管理
Input Validation → 擋掉不合法的資料
API Key → 身分驗證與權限
Error Handling → 統一的錯誤格式

完整程式

app.py

import os
from functools import wraps

from flask import Flask, request
from flask_restx import Api, Resource, fields
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from sqlalchemy.exc import IntegrityError
from werkzeug.exceptions import HTTPException

# ---------- 設定 ----------

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///tasks.db'
app.config['ERROR_404_HELP'] = False

db = SQLAlchemy(app)
migrate = Migrate(app, db, render_as_batch=True)

# ---------- 金鑰與身分 ----------

API_KEYS = {
    os.environ.get('USER_API_KEY', 'user-key-12345'): 'user',
    os.environ.get('ADMIN_API_KEY', 'admin-key-67890'): 'admin',
}

authorizations = {
    'apikey': {
        'type': 'apiKey',
        'in': 'header',
        'name': 'X-API-KEY',
    }
}

api = Api(
    app,
    title='Task API',
    version='1.0',
    description='30 天鐵人賽練習用的作業管理 API',
    authorizations=authorizations,
)

def require_api_key(role=None):
    """檢查 X-API-KEY，role 可以指定需要的身分"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            key = request.headers.get('X-API-KEY')

            if not key or key not in API_KEYS:
                api.abort(401, "請提供有效的 X-API-KEY")

            if role and API_KEYS[key] != role:
                api.abort(403, "你沒有權限執行這個操作")

            return func(*args, **kwargs)
        return wrapper
    return decorator

# ---------- Model（資料表） ----------

class Task(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)
    completed = db.Column(db.Boolean, nullable=False, default=False)
    due_date = db.Column(db.String(20))
    priority = db.Column(db.Integer, nullable=False, default=0)

# ---------- api.model（API 的輸入輸出） ----------

task_input = api.model('TaskInput', {
    'name': fields.String(required=True, description='作業名稱', example='洗碗'),
    'completed': fields.Boolean(required=True, description='是否完成', example=False),
    'due_date': fields.String(description='截止日期 YYYY-MM-DD', example='2026-10-15'),
    'priority': fields.Integer(description='優先度，數字越大越優先', example=0),
})

task_response = api.model('TaskResponse', {
    'id': fields.Integer(readonly=True, description='作業編號'),
    'name': fields.String(description='作業名稱'),
    'completed': fields.Boolean(description='是否完成'),
    'due_date': fields.String(description='截止日期'),
    'priority': fields.Integer(description='優先度'),
})

# ---------- 共用函式 ----------

def find_task(task_id):
    """找一筆作業，找不到就回 404"""
    task = db.session.get(Task, task_id)
    if task is None:
        api.abort(404, f"找不到 id 為 {task_id} 的作業")
    return task

def validate_task_data(data):
    """檢查輸入資料，回傳整理過的資料"""
    name = data.get("name")

    if not isinstance(name, str) or not name.strip():
        api.abort(400, "name 不可以是空白")

    name = name.strip()

    if len(name) > 100:
        api.abort(400, "name 最多只能 100 個字")

    completed = data.get("completed", False)
    if not isinstance(completed, bool):
        api.abort(400, "completed 必須是 true 或 false")

    due_date = data.get("due_date")
    if due_date is not None and not isinstance(due_date, str):
        api.abort(400, "due_date 必須是字串，格式為 YYYY-MM-DD")

    priority = data.get("priority", 0)
    if not isinstance(priority, int) or isinstance(priority, bool):
        api.abort(400, "priority 必須是整數")

    if priority < 0:
        api.abort(400, "priority 不可以小於 0")

    return {
        "name": name,
        "completed": completed,
        "due_date": due_date,
        "priority": priority,
    }

# ---------- 錯誤處理 ----------

@api.errorhandler(IntegrityError)
def handle_integrity_error(error):
    """資料不符合資料庫規則"""
    db.session.rollback()
    return {"message": "資料不符合規則，請檢查輸入的內容"}, 400

@api.errorhandler(Exception)
def handle_unexpected_error(error):
    """未預期的錯誤"""
    if isinstance(error, HTTPException):
        return {"message": error.description}, error.code

    db.session.rollback()
    app.logger.exception("未預期的錯誤")
    return {"message": "伺服器發生錯誤，請稍後再試"}, 500

# ---------- API ----------

@api.route('/tasks')
class TaskList(Resource):

    @api.param('completed', '篩選是否完成，填 true 或 false')
    @api.marshal_list_with(task_response)
    def get(self):
        """取得所有作業"""
        stmt = db.select(Task)

        completed = request.args.get('completed')
        if completed is not None:
            stmt = stmt.filter_by(completed=(completed.lower() == 'true'))

        return db.session.scalars(stmt).all()

    @api.doc(security='apikey')
    @api.expect(task_input, validate=True)
    @api.marshal_with(task_response, code=201)
    @api.response(400, '輸入資料有問題')
    @api.response(401, '沒有提供有效的金鑰')
    @require_api_key()
    def post(self):
        """新增一筆作業"""
        data = validate_task_data(api.payload)

        task = Task(
            name=data["name"],
            completed=data["completed"],
            due_date=data["due_date"],
            priority=data["priority"],
        )

        try:
            db.session.add(task)
            db.session.commit()
        except IntegrityError:
            db.session.rollback()
            api.abort(400, "資料不符合規則，請檢查輸入的內容")

        return task, 201

@api.route('/tasks/<int:task_id>')
@api.param('task_id', '作業編號')
@api.response(404, '找不到作業')
class TaskDetail(Resource):

    @api.marshal_with(task_response)
    def get(self, task_id):
        """取得指定編號的作業"""
        return find_task(task_id)

    @api.doc(security='apikey')
    @api.expect(task_input, validate=True)
    @api.marshal_with(task_response)
    @api.response(400, '輸入資料有問題')
    @api.response(401, '沒有提供有效的金鑰')
    @require_api_key()
    def put(self, task_id):
        """修改指定編號的作業"""
        task = find_task(task_id)
        data = validate_task_data(api.payload)

        task.name = data["name"]
        task.completed = data["completed"]
        task.due_date = data["due_date"]
        task.priority = data["priority"]

        db.session.commit()
        return task

    @api.doc(security='apikey')
    @api.response(204, '刪除成功')
    @api.response(403, '需要管理員權限')
    @require_api_key(role='admin')
    def delete(self, task_id):
        """刪除指定編號的作業（只有管理員可以）"""
        task = find_task(task_id)

        db.session.delete(task)
        db.session.commit()

        return '', 204

if __name__ == '__main__':
    app.run(debug=True)

怎麼從零跑起來

  1. 建立虛擬環境並啟動
python -m venv .venv
.venv\Scripts\Activate.ps1
  1. 安裝套件
pip install Flask==3.1.3 flask-restx==1.3.2 Flask-SQLAlchemy==3.1.1 Flask-Migrate==4.1.0

也可以寫成 requirements.txt：

Flask==3.1.3
flask-restx==1.3.2
Flask-SQLAlchemy==3.1.1
Flask-Migrate==4.1.0

然後用：

pip install -r requirements.txt
  1. 建立資料庫

因為已經改用 Flask-Migrate 管理（Day 23），所以不是 create_all()，而是：

flask db init
flask db migrate -m "create task table"
flask db upgrade
  1. 啟動
python app.py
  1. 打開 Swagger

http://127.0.0.1:5000/

https://ithelp.ithome.com.tw/upload/images/20261013/20183865G3qqjVx1uw.png

完整版的 Swagger：五個 API、中文說明、右上角 Authorize、需要金鑰的旁邊有鎖頭

完整跑一次

把前面學過的東西全部驗證一次：

  1. GET /tasks → 200
  2. 不帶金鑰 POST /tasks → 401
  3. 按 Authorize 輸入 user-key-12345，再 POST：
{
  "name": "寫 Day 29",
  "completed": false,
  "due_date": "2026-10-15",
  "priority": 2
}

→ 201

  1. POST 一筆不合法的 {"name": " ","completed": false} → 400「name 不可以是空白」
  2. GET /tasks?completed=false → 只會查到未完成的
  3. PUT /tasks/1 → 200（注意 PUT 是整筆取代，沒傳的欄位會被清成 null，Day 15 講過）
  4. 用 user 金鑰 DELETE → 403
  5. 換成 admin-key-67890 再刪 → 204
  6. GET /tasks/999 → 404「找不到 id 為 999 的作業」
  7. 關掉 Server 再重開，資料還在

這十個情境前面幾天都分別測過了，今天是確認它們湊在一起也沒有互相打架。

每一段是哪一天來的？

把程式對照回前面幾天，其實整份 app.py 就是這 29 天的整理：
Flask、@app.route → Day 7、Day 8
回傳 JSON → Day 9
Api、Resource、@api.route → Day 11
Swagger → Day 12
api.model、@api.expect、api.payload → Day 13
CRUD、路徑參數、api.abort → Day 14、Day 15
marshal、docstring 文件 → Day 16
資料庫、ORM → Day 17～Day 19
Model、db.session → Day 20～Day 22
completed 篩選（request.args）→ Day 22 的小延伸
Migration、server_default → Day 23
錯誤處理、狀態碼、ERROR_404_HELP → Day 24
Input Validation、白名單 → Day 25
API Key、401/403、@wraps → Day 26
參數化查詢（ORM 幫我們做） → Day 27、Day 28

每一個 decorator 背後都是一天的內容，現在看起來就比較有感覺了。

如果想再整理下去

現在整個程式都寫在一個 app.py 裡，兩百多行了。

實際的專案通常會拆開，大概會像這樣：

project/
├── app.py # 建立 app、註冊設定和 API
├── config.py        # 設定
├── extensions.py    # db = SQLAlchemy()、migrate = Migrate()
├── models.py        # Task 等 Model
└── apis/
    └── task.py      # Task 的 Resource

Flask-RESTX 有提供 Namespace，可以把一組相關的 API 收在一起：

# apis/task.py
from flask_restx import Namespace, Resource

ns = Namespace('tasks', description='作業相關的 API')

@ns.route('')
class TaskList(Resource):
    ...

# app.py
api.add_namespace(ns, path='/tasks')

這樣 Swagger 上也會依照 Namespace 分組，API 多的時候看起來會清楚很多。

這部分就先不展開了，有興趣可以自己試試看。

今天把 30 天學到的東西全部放進同一支程式，完成了一個具備這些能力的 API：

有 API 文件可以直接測試
資料真的存在資料庫，重開不會不見
資料表結構的改變有 migration 紀錄
會檢查輸入的資料
有身分驗證和權限區分
出錯的時候會回傳正確的狀態碼和看得懂的訊息

明天就是最後一天了，來做個總回顧，也聊聊這 30 天遇到的問題和心得。


上一篇
Day 28｜怎麼避免 SQL Injection？
系列文
從零開始的後端開發：用 Flask 實作 REST API，搞懂 API 與資料庫之間如何協作 共 29 篇
  1. 25
    Day 25｜使用者亂傳資料怎麼辦？
  2. 26
    Day 26｜API 安全入門
  3. 27
    Day 27｜ SQL Injection 是什麼？
  4. 28
    Day 28｜怎麼避免 SQL Injection？
  5. 29
    Day 29｜把全部串起來
完整目錄
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言
iThome鐵人賽
參賽組數
929 組
團體組數
32 組
累計文章數
19622 篇
完賽人數
317 人
iT+看影片追技術 看影片追技術 看更多
熱門tag
15th鐵人賽 16th鐵人賽 13th鐵人賽 17th鐵人賽 14th鐵人賽 18th鐵人賽 12th鐵人賽 11th鐵人賽 鐵人賽 2019鐵人賽 javascript 2018鐵人賽 python 2017鐵人賽 windows php c# linux windows server css
熱門問題
熱門回答
熱門文章
IT邦幫忙