エラーハンドリング

API でエラーが発生した際に以上終了したい場合は、werkzeug.exceptions から BadRequest などをインポートして処理を中止したいタイミングに raise すれば良い。

from werkzeug.exceptions import BadRequest, Unauthorized, NotFound, InternalServerError

@app.post("/sample")
def sample():
    error = True
    if error:
        raise BadRequest
    else:
        return render_template('sample.html')

これでも一応エラーハンドリングできるわけだが、画面はすでに用意されたものを使うしかないので融通が効かない。 これをカスタマイズするには、app にエラーハンドリング用のメソッドを紐づける必要がある。

app = Flask(__name__)

# 省略

def handle_bad_request(e):
    return render_template('error.html', e=e), 400

def handle_unauthorized(e):
    return render_template('error.html', e=e), 401

def handle_not_found(e):
    return render_template('error.html', e=e), 404

def handle_internal_server_error(e):
    return render_template('error.html', e=e), 500

app.register_error_handler(400, handle_bad_request)
app.register_error_handler(401, handle_unauthorized)
app.register_error_handler(404, handle_not_found)
app.register_error_handler(500, handle_internal_server_error)

このように書いてあげれば、raise BadRequest 時には handle_bad_request が呼び出され、raise InternalServerError 時には handle_internal_server_error が呼び出されるようになる。 そうすると、error.html が呼び出され、そのテンプレートにエラーの内容を含む変数 e が渡るので、それを画面に表示することができる。

<!-- error.html -->
{% extends 'base.html' %}

{% block header %}
  <a href="/"><h1>{% block title %}Error{% endblock %}</h1></a>
{% endblock %}

{% block content %}
  <section>
    <h1>Something went wrong.</h1>
    <p>{{ e }}</p>
  </section>
{% endblock %}

また、API を try except でラップしてその他のエラーを補足することもできるので、必要なら取り入れてもよさそう。 例えば sign_up() なら以下のような感じ。

@app.route("/sign_up", methods=('GET', 'POST'))
def sign_up():
    try:
        if 'email' in session:
            return redirect(url_for('products'))

        form = SignUpForm(request.form)
        if form.validate_on_submit():
            name = form.name.data
            email = form.email.data
            password = form.password.data
            salt = bcrypt.gensalt(rounds=12, prefix=b'2b')
            hashpw = bcrypt.hashpw(password.encode('utf8'), salt)
            user = User(name=name, email=email, password=hashpw.decode('utf8'))
            db.session.add(user)
            db.session.commit()
            session['email'] = user.email
            return redirect(url_for('products'))

        return render_template('sign_up.html', form=form)
    except Exception as e:
        return render_template('sign_up.html', form=form, e=e)

この場合、sign_up.html テンプレートで e を表示するように修正する。

<!-- sign_up.html -->
<form method="post">
  {{ form.csrf_token }}
  <dl>
    {{ render_field(form.name, placeholder="name") }}
    {{ render_field(form.email, placeholder="email") }}
    {{ render_field(form.password, placeholder="password") }}
  </dl>
  <input type="submit" value="サインアップ">
  <p>{{ e }}</p> <!-- ここで　　e を表示している -->
</form>

pytest

テストを行う前に、これまでに作ってきたディレクトリの構成を整理したい。 で、テストでも API を呼び出す必要があるので、flask app を生成する create_app() を作り、これをテストファイルから呼び出すようにする。

ちょっと変更が多めなので詳しくはリポジトリを確認してほしいが、これまで main.py に書いていたコードを __init__.py に移動し、これを create_app() でまとめて main.py からそれを呼び出し、app として定義している。

# __init__.py
def create_app():
    app = Flask(__name__)

    # 省略

    if os.environ['FLASK_ENV'] == 'development':
        app.config["SQLALCHEMY_DATABASE_URI"] = 'postgresql://postgres:postgres@postgres/development'
    else:
        app.config["TESTING"] = True
        app.config["WTF_CSRF_ENABLED"] = False
        app.config["SQLALCHEMY_DATABASE_URI"] = 'postgresql://postgres:postgres@postgres/test'

    # 省略

    from . import views
    app.register_blueprint(views.bp)

    return app

# main.py
from . import create_app

app = create_app()

変わったところとしては、環境変数ごとに DB を切り替えたのと、API を別ファイル(views.py)へ移動したので、views 側で BluePrint として app(bp) を呼び出せるようにしておく。 views.py からは bp = Blueprint("views", __name__) で app を参照し、@bp.route("/sign_up", methods=('GET', 'POST')) みたいに使う。

# views.py
bp = Blueprint("views", __name__)

@bp.route("/sign_up", methods=('GET', 'POST'))
# 省略

フォームとモデルもそれぞれ別のファイルとして切り出し、views から参照するようにしている。

で、また __init__.py の話に戻るが、データベースの定義を開発環境とテスト環境でわけている。 データベースの作成は PostgreSQL コンテナ作成時に仕込んだ docker-entrypoint-initdb.d の SQL ファイルから行っている。 docker-compose.yml の postgres は以下のように volumes で /db/initdb.d を読み込んでいるが、このディレクトリの中に SQL ファイルが入っている。

postgres:
  image: postgres:16.2-bookworm
  environment:
    - POSTGRES_PASSWORD=postgres
    - POSTGRES_USER=postgres
    - POSTGRES_DB=postgres
    - TZ=Asia/Tokyo
  ports:
    - 5432:5432
  volumes:
    - ./db/data:/var/lib/postgresql/data
    - ./db/initdb.d:/docker-entrypoint-initdb.d

SQL ファイルには以下のような内容が記載されている。 docker-entrypoint-initdb.d の中の SQL ファイルはコンテナ起動時に実行されるので、PostgreSQL が立ち上がった時点でこれらのデータベースが作成される。

CREATE DATABASE development;
CREATE DATABASE test;

あとは docker-compose.yml で開発環境とテスト環境のコンテナをわけ、それぞれの environment に FLASK_ENV を設定する。 これが上述の __init__.py の条件分岐で参照されることになる。

dev_app:
  build: .
  volumes:
    - .:/app
  ports:
    - 5001:5000
  command: bash -c "flask run --host=0.0.0.0"
  environment:
    - FLASK_APP=main
    - FLASK_DEBUG=on
    - FLASK_ENV=development
    - TZ=Asia/Tokyo
test_app:
  build: .
  volumes:
    - .:/app
  command: bash -c "flask run --host=0.0.0.0"
  environment:
    - FLASK_APP=main
    - FLASK_DEBUG=on
    - FLASK_ENV=test
    - TZ=Asia/Tokyo

再度コンテナを立ち上げて、開発環境がこれまで同様に起動すれば問題ない。 テスト環境の方へ入ってテストを進めていく。

まず、tests ディレクトリを最上位のディレクトリに作成する。 その中に __init__.py と conftest.py を作成する。 __init__.py には何も記載する必要はないが、conftest.py には以下を書く。

import pytest
from .. import create_app, db

@pytest.fixture()
def app():
    app = create_app()

    # set up
    with app.app_context():
        db.create_all()

    yield app

    # tear down
    with app.app_context():
        db.drop_all()

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

@pytest.fixture()
def runner(app):
    return app.test_cli_runner()

このファイルはテストの初期化を行う。 set up コメントの部分でデータを用意し、tear down コメントの部分で全てのデータをドロップする。

そして、実際のテストは conftest.py ディレクトリと同じ階層に test プリフィックスで作成したファイルの中に書いていく。

とりあえず、サンプルとしてサインアップ機能のテストは以下のような感じになるかと思う。

from flask import session
from .. import db
import bcrypt
from ..models import User

def test_sign_up_get(client):
    with client:
        response = client.get('/sign_up')
        assert response.status_code == 200

def test_sign_up_post(client):
    name = "test"
    email = "[email protected]"
    password = "hogehoge"

    with client:
        response = client.post('/sign_up', data={
            "name": name,
            "email": email,
            "password": password
        })
        assert response.status_code == 302
        assert session.get('email') == email
        user = db.session.execute(db.select(User).filter_by(email=email)).scalar_one_or_none()
        assert user.name == name
        assert user.email == email
        assert bcrypt.checkpw(password.encode('utf8'), user.password.encode('utf8'))

def test_sign_up_user_already_exists(app, client):
    name = "test"
    email = "[email protected]"
    password = "hogehoge"

    with app.app_context():
        salt = bcrypt.gensalt(rounds=12, prefix=b'2b')
        hashpw = bcrypt.hashpw(password.encode('utf8'), salt)
        data = User(name=name, email=email, password=hashpw.decode('utf8'))
        db.session.add(data)
        db.session.commit()

    with client:
        response = client.post('/sign_up', data={
            "name": name,
            "email": email,
            "password": password
        })
        assert response.status_code == 200
        assert session.get('email') is None

リポジトリの方には一応、サインインとサインアウトのテストも作っているので、必要なら参考にしてみてください。 create_app() からのくだりから flask の面倒臭さが垣間見える気がするが、慣れの問題かとも思う。 とりあえず、テストについてもこれで終わり。

を仕舞い

Flask で Web アプリケーション、特に EC サイトをイメージして作ってみた。 UI とかは素のままで何の色気もないが、おおまかな機能については触れたつもりだ。 まだまだ奥は深く、これで Web アプリケーションの構築を知った気持ちになってはならないが、導入としては読めるかも？と思いたい。

Flask は今後は使わないかもだが FastAPI とかでまた何か書くかもです。