0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@cozy_666_techin株式会社Flexibility

Flaskのブループリント — アプリが大きくなってきたら

0
Posted at

はじめに

Flask入門の記事でブループリントを少し触れたが、もう少し深掘りした。

最初はapp.py1ファイルに全部書いていたが、エンドポイントが増えてくると管理しきれなくなってきた。LaravelのRouteグループとコントローラーの分離に相当することをFlaskでやろうとするとBlueprintを使うことになる。

Application Factoryパターンと組み合わせることで、テストもしやすい構成になった。

ブループリントなしの問題

まず整理する。小さいうちはこれでいい。

# app.py（全部1ファイル）
from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route("/users")
def user_list():
    return jsonify([{"id": 1}])

@app.route("/users/<int:user_id>")
def user_detail(user_id):
    return jsonify({"id": user_id})

@app.route("/posts")
def post_list():
    return jsonify([{"id": 1}])

@app.route("/posts/<int:post_id>")
def post_detail(post_id):
    return jsonify({"id": post_id})

# ... エンドポイントが増えるほど肥大化する

エンドポイントが10本を超えてくるとこのファイルが数百行になる。Laravelがルートファイルとコントローラーを分けているように、Flaskもファイルを分割したい。

Blueprint基本

# blueprints/users.py
from flask import Blueprint, jsonify, request

users_bp = Blueprint(
    "users",           # Blueprintの名前（エンドポイント名のプレフィックスになる）
    __name__,
    url_prefix="/users",
)

@users_bp.route("/")
def user_list():
    return jsonify([{"id": 1, "name": "田中"}])

@users_bp.route("/<int:user_id>")
def user_detail(user_id):
    return jsonify({"id": user_id})

@users_bp.route("/", methods=["POST"])
def user_create():
    data = request.get_json()
    return jsonify({"id": 99, "name": data["name"]}), 201

# app.py
from flask              import Flask
from blueprints.users   import users_bp
from blueprints.posts   import posts_bp

app = Flask(__name__)
app.register_blueprint(users_bp)
app.register_blueprint(posts_bp)

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

register_blueprint()でアプリにBlueprintを登録する。url_prefixはBlueprintに設定する方法と、register_blueprint()に渡す方法の2通りある。

# register_blueprint側でprefixを指定する方法
app.register_blueprint(users_bp, url_prefix="/api/v1/users")

Application Factoryパターン

Blueprintと一緒に覚えておきたいのがApplication Factory。

Factoryなしの問題

# app.pyでappをグローバルに作る（よくある問題のある書き方）
app = Flask(__name__)

# テストのときに設定を差し替えにくい
# 循環importが起きやすい

Application Factoryを使う

# app.py
from flask import Flask

def create_app(config_name: str = "development") -> Flask:
    app = Flask(__name__)

    # 設定を読み込む
    app.config.from_object(f"config.{config_name.capitalize()}Config")

    # Blueprintを登録
    from blueprints.users  import users_bp
    from blueprints.posts  import posts_bp
    from blueprints.health import health_bp

    app.register_blueprint(users_bp)
    app.register_blueprint(posts_bp)
    app.register_blueprint(health_bp)

    # 拡張機能の初期化
    from extensions import db, migrate, jwt
    db.init_app(app)
    migrate.init_app(app, db)
    jwt.init_app(app)

    return app

# wsgi.py（本番エントリーポイント）
from app import create_app

app = create_app("production")

# run.py（開発エントリーポイント）
from app import create_app

app = create_app("development")

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

LaravelのServiceProviderでアプリを初期化するのに近い発想。create_app()に設定名を渡せるのでテスト時に別の設定を使いやすい。

設定ファイルの分割

# config.py
import os

class BaseConfig:
    SECRET_KEY          = os.environ.get("SECRET_KEY", "dev-secret")
    SQLALCHEMY_TRACK_MODIFICATIONS = False

class DevelopmentConfig(BaseConfig):
    DEBUG               = True
    SQLALCHEMY_DATABASE_URI = os.environ.get(
        "DATABASE_URL", "sqlite:///dev.db"
    )

class ProductionConfig(BaseConfig):
    DEBUG               = False
    SQLALCHEMY_DATABASE_URI = os.environ.get("DATABASE_URL")
    # 本番用セキュリティ設定
    SESSION_COOKIE_SECURE = True
    SESSION_COOKIE_HTTPONLY = True

class TestingConfig(BaseConfig):
    TESTING             = True
    SQLALCHEMY_DATABASE_URI = "sqlite:///:memory:"  # インメモリDB
    WTF_CSRF_ENABLED    = False

拡張機能の管理

Flask-SQLAlchemy、Flask-Migrate、Flask-JWTなどの拡張機能をApplication Factoryと組み合わせるときはextensions.pyでインスタンスを作っておく。

# extensions.py
from flask_sqlalchemy   import SQLAlchemy
from flask_migrate      import Migrate
from flask_jwt_extended import JWTManager

db      = SQLAlchemy()
migrate = Migrate()
jwt     = JWTManager()

# app.py
from extensions import db, migrate, jwt

def create_app(config_name: str = "development") -> Flask:
    app = Flask(__name__)
    app.config.from_object(f"config.{config_name.capitalize()}Config")

    # 拡張機能にappを渡して初期化
    db.init_app(app)
    migrate.init_app(app, db)
    jwt.init_app(app)

    # Blueprintを登録
    ...

    return app

拡張機能をextensions.pyで一元管理することで循環importを避けられる。

プロジェクト構成

ここまでの内容をまとめた構成。

myapp/
├── app.py              # create_app()
├── wsgi.py             # 本番エントリーポイント
├── run.py              # 開発エントリーポイント
├── config.py           # 設定クラス
├── extensions.py       # Flask拡張機能のインスタンス
├── models/
│   ├── __init__.py
│   ├── user.py
│   └── post.py
├── blueprints/
│   ├── __init__.py
│   ├── users/
│   │   ├── __init__.py
│   │   ├── routes.py   # Blueprintの定義
│   │   ├── schemas.py  # marshmallowスキーマ
│   │   └── service.py  # ビジネスロジック
│   ├── posts/
│   │   ├── __init__.py
│   │   ├── routes.py
│   │   └── service.py
│   └── health/
│       ├── __init__.py
│       └── routes.py
└── tests/
    ├── conftest.py
    ├── test_users.py
    └── test_posts.py

機能ごとにディレクトリを切ってその中にroutes・schemas・serviceを置くパターン。LaravelのApp/Http/Controllers + App/Services構造に近い。

Blueprintの詳細な書き方

# blueprints/users/routes.py
from flask              import Blueprint, jsonify, request, abort
from models.user        import User
from extensions         import db
from .schemas           import UserSchema

users_bp = Blueprint("users", __name__, url_prefix="/users")
schema   = UserSchema()
schemas  = UserSchema(many=True)

@users_bp.route("/")
def user_list():
    page     = request.args.get("page", 1, type=int)
    per_page = request.args.get("per_page", 20, type=int)

    pagination = User.query.paginate(page=page, per_page=per_page)
    return jsonify({
        "data":  schemas.dump(pagination.items),
        "total": pagination.total,
        "page":  pagination.page,
        "pages": pagination.pages,
    })

@users_bp.route("/<int:user_id>")
def user_detail(user_id):
    user = User.query.get_or_404(user_id)
    return jsonify(schema.dump(user))

@users_bp.route("/", methods=["POST"])
def user_create():
    data   = request.get_json()
    errors = schema.validate(data)
    if errors:
        return jsonify({"errors": errors}), 422

    user = User(**schema.load(data))
    db.session.add(user)
    db.session.commit()
    return jsonify(schema.dump(user)), 201

@users_bp.route("/<int:user_id>", methods=["PUT"])
def user_update(user_id):
    user   = User.query.get_or_404(user_id)
    data   = request.get_json()
    errors = schema.validate(data, partial=True)
    if errors:
        return jsonify({"errors": errors}), 422

    for key, value in schema.load(data, partial=True).items():
        setattr(user, key, value)
    db.session.commit()
    return jsonify(schema.dump(user))

@users_bp.route("/<int:user_id>", methods=["DELETE"])
def user_delete(user_id):
    user = User.query.get_or_404(user_id)
    db.session.delete(user)
    db.session.commit()
    return "", 204

get_or_404()はIDで検索して見つからなければ自動で404を返す。LaravelのfindOrFail()に相当してよく使う。

エラーハンドラーをBlueprintに登録

# blueprints/errors.py
from flask import Blueprint, jsonify
from werkzeug.exceptions import HTTPException

errors_bp = Blueprint("errors", __name__)

@errors_bp.app_errorhandler(400)
def bad_request(e):
    return jsonify({"error": "リクエストが不正です", "detail": str(e)}), 400

@errors_bp.app_errorhandler(401)
def unauthorized(e):
    return jsonify({"error": "認証が必要です"}), 401

@errors_bp.app_errorhandler(403)
def forbidden(e):
    return jsonify({"error": "アクセス権がありません"}), 403

@errors_bp.app_errorhandler(404)
def not_found(e):
    return jsonify({"error": "リソースが見つかりません"}), 404

@errors_bp.app_errorhandler(422)
def unprocessable(e):
    return jsonify({"error": "バリデーションエラー", "detail": str(e)}), 422

@errors_bp.app_errorhandler(500)
def server_error(e):
    return jsonify({"error": "サーバーエラーが発生しました"}), 500

# HTTPException全般をキャッチ
@errors_bp.app_errorhandler(HTTPException)
def handle_http_exception(e):
    return jsonify({"error": e.description}), e.code

エラーハンドラーを専用のBlueprintに切り出すとapp.pyがすっきりする。@errors_bp.app_errorhandler()はアプリ全体に適用される（errorhandlerではなくapp_errorhandlerを使う）。

Blueprintでのテスト

Application Factoryを使うとテストで設定を差し替えやすい。

# tests/conftest.py
import pytest
from app      import create_app
from extensions import db as _db

@pytest.fixture(scope="session")
def app():
    app = create_app("testing")  # テスト用設定
    with app.app_context():
        _db.create_all()
        yield app
        _db.drop_all()

@pytest.fixture
def client(app):
    return app.test_client()

@pytest.fixture
def db(app):
    with app.app_context():
        yield _db
        _db.session.rollback()  # テストごとにロールバック

# tests/test_users.py
import json

def test_user_list(client):
    response = client.get("/users/")
    assert response.status_code == 200
    data = response.get_json()
    assert isinstance(data["data"], list)

def test_user_create(client, db):
    response = client.post(
        "/users/",
        data=json.dumps({"name": "田中", "email": "tanaka@example.com"}),
        content_type="application/json",
    )
    assert response.status_code == 201
    data = response.get_json()
    assert data["name"] == "田中"

def test_user_not_found(client):
    response = client.get("/users/99999")
    assert response.status_code == 404

FastAPIとの比較

Blueprint構成のFlaskとFastAPIのルーター構成を並べると似ている。

# Flask Blueprint
from flask import Blueprint

users_bp = Blueprint("users", __name__, url_prefix="/users")

@users_bp.route("/")
def user_list(): ...

# main app
app.register_blueprint(users_bp)

# FastAPI Router
from fastapi import APIRouter

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/")
def user_list(): ...

# main app
app.include_router(router)

BlueprintがAPIRouterに、register_blueprint()がinclude_router()に対応する。FastAPIのほうがHTTPメソッドをデコレータで明示する分、見た目でメソッドが一目でわかりやすい。

まとめ

  • Blueprintでルートをファイルに分割する
  • Application Factoryでcreate_app()を作るとテストしやすくなる
  • 拡張機能はextensions.pyに集めて循環importを防ぐ
  • エラーハンドラーを専用Blueprintに切り出すとapp.pyがすっきりする
  • get_or_404()でfindOrFail()相当の処理を簡潔に書ける

Flaskは小さく始めて大きくなったら分割するという設計が自然にできる。Application Factoryパターンを最初から採用しておくと、後からテスト設定を追加するときに楽になる。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?