こんにちは、かつコーチです。
ここまでのシリーズで、ルーティング・テンプレート・DB連携・認証・テスト・デプロイと一通り扱ってきました。
最終回となる今回は、これまでの知識を組み合わせて、シンプルな家計簿アプリのREST APIを実際に構築します。
本記事は上級者向けとして、Flask本体の基礎説明は省略し、実装とアーキテクチャ判断に絞って解説します。
設計方針とプロジェクト構成
今回作るAPIの仕様
家計簿の「支出(Expense)」を管理するシンプルなREST APIを想定します。
| メソッド | パス | 用途 |
|---|---|---|
| GET | /api/expenses | 支出一覧の取得 |
| POST | /api/expenses | 支出の登録 |
| GET | /api/expenses/ | 支出の詳細取得 |
| PUT | /api/expenses/ | 支出の更新 |
| DELETE | /api/expenses/ | 支出の削除 |
認証はFlask-Loginで扱ったセッション方式ではなく、API向けにトークン認証(リクエストヘッダーにトークンを付与して本人確認する方式)を採用します。
REST APIはステートレス(サーバー側にログイン状態を保持しない)が原則のため、セッションベースの認証とは相性が悪い点が理由です。
Blueprintでファイルを分割する
小規模でも、後から拡張しやすいようBlueprintでAPI部分を分離しておきます。
kakeibo_api/
├── app.py
├── models.py
├── schemas.py
└── api/
├── __init__.py
└── expenses.py
# api/expenses.py
from flask import Blueprint, request, jsonify
from models import db, Expense
expenses_bp = Blueprint("expenses", __name__, url_prefix="/api/expenses")
app.py側ではapp.register_blueprint(expenses_bp)と登録するだけで済みます。
機能が増えてもapp.pyが肥大化しないため、実務でも標準的に採用されている構成です。
実装:CRUD APIとバリデーション
モデル定義とマーシャリング
# models.py
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime
db = SQLAlchemy()
class Expense(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(100), nullable=False)
amount = db.Column(db.Integer, nullable=False)
category = db.Column(db.String(50), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
def to_dict(self):
return {
"id": self.id,
"title": self.title,
"amount": self.amount,
"category": self.category,
"created_at": self.created_at.isoformat(),
}
to_dictメソッドをモデル自身に持たせておくことで、マーシャリング(PythonオブジェクトをJSONで返せる形式に変換する処理)のロジックが各エンドポイントに散らばるのを防げます。
バリデーションを一元化する
リクエストの値チェックをエンドポイントごとにベタ書きすると、ルートが増えるにつれてコードの重複が増えます。
def validate_expense_payload(data):
errors = []
if not data.get("title"):
errors.append("titleは必須です")
if not isinstance(data.get("amount"), int) or data.get("amount") <= 0:
errors.append("amountは正の整数で指定してください")
if data.get("category") not in ["食費", "交通費", "光熱費", "娯楽費", "その他"]:
errors.append("categoryが不正な値です")
return errors
@expenses_bp.route("", methods=["POST"])
def create_expense():
data = request.get_json(silent=True) or {}
errors = validate_expense_payload(data)
if errors:
return jsonify({"errors": errors}), 400
expense = Expense(title=data["title"], amount=data["amount"], category=data["category"])
db.session.add(expense)
db.session.commit()
return jsonify(expense.to_dict()), 201
request.get_json(silent=True)とすることで、JSON形式でないリクエストが来た場合に例外を発生させずNoneを返させ、その後のor {}で安全に空辞書として扱えます。
一覧取得とページネーション
家計簿のようにデータが増え続けるリソースでは、全件返却は避け、ページネーション(データを分割して少しずつ返す仕組み)を最初から組み込んでおくのが実務での定石です。
@expenses_bp.route("", methods=["GET"])
def list_expenses():
page = request.args.get("page", 1, type=int)
per_page = request.args.get("per_page", 20, type=int)
pagination = Expense.query.order_by(Expense.created_at.desc()).paginate(
page=page, per_page=per_page, error_out=False
)
return jsonify({
"items": [e.to_dict() for e in pagination.items],
"total": pagination.total,
"page": pagination.page,
"pages": pagination.pages,
})
paginateはFlask-SQLAlchemyに標準で用意されており、error_out=Falseを指定することで、存在しないページ番号を指定された場合に例外ではなく空配列を返すようにできます。
エラーハンドリングとレスポンス設計
統一されたエラーレスポンス形式
エンドポイントごとにエラーレスポンスの形式がバラバラだと、フロントエンド側の実装が複雑になります。
from flask import Flask, jsonify
from werkzeug.exceptions import HTTPException
app = Flask(__name__)
@app.errorhandler(HTTPException)
def handle_http_exception(e):
return jsonify({"error": e.name, "message": e.description}), e.code
@app.errorhandler(404)
def handle_not_found(e):
return jsonify({"error": "Not Found", "message": "指定されたリソースが見つかりません"}), 404
app.errorhandlerをアプリ全体に登録しておくことで、abort(404)やget_or_404が発生させる例外も、一律JSON形式で返せるようになります。
これを設定し忘れると、APIなのにFlaskのデフォルトのHTML形式エラーページが返ってしまい、フロントエンド側でJSONパースエラーが起きるという事故につながります。
実際に私も初めてAPIを作った際、404のときだけHTMLが返ってきてフロントのresponse.json()が例外を吐く、というトラブルに遭遇したことがあります。
トークン認証の簡易実装
本格的な認証にはFlask-JWT-Extendedのようなライブラリを使うのが一般的ですが、仕組みの理解のためにシンプルな実装を示します。
from functools import wraps
from flask import request, jsonify
API_TOKEN = "your-static-api-token" # 実運用では環境変数管理
def token_required(f):
@wraps(f)
def decorated(*args, **kwargs):
auth_header = request.headers.get("Authorization", "")
if auth_header != f"Bearer {API_TOKEN}":
return jsonify({"error": "Unauthorized"}), 401
return f(*args, **kwargs)
return decorated
@expenses_bp.route("", methods=["GET"])
@token_required
def list_expenses():
...
固定トークンでの実装はあくまで学習目的です。
複数ユーザーが利用する本番のAPIでは、有効期限付きのJWT(JSON Web Token)を発行する仕組みへの置き換えを検討してください。
応用・一歩先の使い方
CORS対応でフロントエンドと連携する
React・Vueなど別ドメインで動くフロントエンドからこのAPIを呼び出す場合、CORS(Cross-Origin Resource Sharing)の設定が必要になります。
pip install flask-cors
from flask_cors import CORS
CORS(app, resources={r"/api/*": {"origins": "https://your-frontend.example.com"}})
originsを*にすると全てのドメインからのアクセスを許可してしまうため、本番運用では許可するドメインを明示的に絞り込むことをおすすめします。
OpenAPI(Swagger)でドキュメントを自動生成する
APIが育ってくると、エンドポイントの仕様書を手動で管理するのは現実的ではありません。
Flask-Smorestのようなライブラリを使うと、コードからOpenAPI形式のドキュメントを自動生成し、Swagger UIで確認できるようになります。
チーム開発やフロントエンドとの連携が発生するプロジェクトでは、早い段階での導入を検討する価値があります。
まとめ
この記事のポイント
- REST APIはBlueprintで機能ごとに分割し、モデルに
to_dictを持たせてマーシャリングを一元化する - バリデーションは共通関数化してエンドポイントの重複を減らす
- 一覧APIには最初からページネーションを組み込むのが実務の定石
app.errorhandlerで統一されたJSON形式のエラーレスポンスを返す- セッション認証ではなくトークン認証がREST APIの原則に合う
シリーズを振り返って
F01の入門編から始まり、環境構築、ルーティング、テンプレート、DB連携、認証、テスト、デプロイ、そして今回のREST API実践まで、Flaskの主要な機能を一通り扱ってきました。
DjangoとFlaskの使い分けについては、当ブログの「Django・Flask比較」記事も参考にしてください。
軽量フレームワークとしてのFlaskの強みを活かしつつ、必要な機能を拡張ライブラリで積み上げていく感覚がつかめていれば、このシリーズの目的は達成です。
タグ: Flask, 上級者向け, 実践プロジェクト
