なぜ Flask で作る？

Flask で作るのは最近ちょっと趣味でいじってたからであって特に深い意味はないです。 何を使うにせよ、考え方は特に変わらないと思う。

作るものは簡単な EC サイト。 実際の購入機能を含めるのであれば Stripe を組み込むなどすれば良さそう。 Stripe も本業でよく触ってるので今度カード決済やコンビニ決済の実装について書こうと思う。

今回は DB に登録した商品情報の一覧画面と詳細画面、またその購入（形だけ）と購入履歴の表示、およびショッピングカートの実装を行う。 認証、エラーハンドリング、テストについては別の記事に分けるつもり。

Docker で Flask アプリを起動する

docker 使って作っていきます。 こういう時はまず Dockerfile と docker-compose.yml を作って、コンテナの中で flask アプリの原型を作り、その過程のコマンドをメモするようにしてます。

FROM python:3.12.2-slim-bookworm
WORKDIR /app
RUN apt update -y && pip install --upgrade pip

version: "3.8"
services:
  app:
    build: .
    tty: true

以上のファイルを2つ作って docker compose build と docker compose up -d でコンテナを起動する。 あとは中に入って作業していきます。

まず、flask アプリを作ってみる。

$ docker exec -it flask_web-app-1 /bin/bash
root@f7c40238f45c:/app# pip install Flask
root@f7c40238f45c:/app# touch main.py
root@f7c40238f45c:/app# apt install vim

QuickStart を参考に main.py を作成する。 ついでに vim を入れる。

# main.py
from flask import Flask

app = Flask(__name__)

@app.route("/")
def hello_world():
    return "<p>Hello, World!</p>"

あとは flask アプリを起動する。

root@f7c40238f45c:/app# flask --app main run
 * Serving Flask app 'main'
 * Debug mode: off
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on http://127.0.0.1:5000
Press CTRL+C to quit

5000 ポートで起動しているので、こちらを解放してあげればホスト側からもアクセスできるとわかる。

依存関係は requirements.txt で管理したいので、Dockerfile をいじる。 最近は poetry も良さそう？

イメージは慣れてるので debian で、slim があれば使う。 本番環境で使う時は root 以外の権限で起動したいところだが、今回はその辺は考えない。 この記事では脆弱性があってもイメージを利用するが、気になれば別のイメージを使ったり、マルチステージビルドすれば良い。

FROM python:3.12.2-slim-bookworm
WORKDIR /app
COPY . /app
RUN apt update -y \
  && pip install --upgrade pip \
  && pip install -r requirements.txt

Flask>=3.0,<4.0

version: "3.8"
services:
  app:
    build: .
    volumes:
      - .:/app
    ports:
      - 5001:5000
    command: bash -c "flask --app main run --host=0.0.0.0"
    environment:
      - FLASK_APP=main
      - FLASK_DEBUG=on
      - TZ=Asia/Tokyo

main.py もホスト側で作っておいてディレクトリに入れておく。 ホスト側のポートは 5001 にしておく。 んで、再度コンテナを作り直す。 ちなみに environment に FLASK_APP を指定しておかないと、のちの flask shell でエラーになるのでこのタイミングで指定する。 FLASK_APP を指定すれば、そのモジュールがアプリの起点となるので、起動時に --app を指定する必要もなくなる。 FLASK_DEBUG=on はデバッグモードの設定で、デバッグモードではホットリロードが効くようになるから設定しておく。

$ docker compose build
$ docker up -d
$ docker ps
CONTAINER ID   IMAGE           COMMAND                  CREATED         STATUS         PORTS                    NAMES
d7590bc86543   flask_web-app   "bash -c 'flask --ap…"   6 seconds ago   Up 5 seconds   0.0.0.0:5001->5000/tcp   flask_web-app-1

localhost:5001 を確認して Hello, World! が出てきたら問題なし。 ここからスタート。

よくある CRUD API を作る

データを処理する API を作っていく。 これを進めるためにはまず、データベースの設定とモデルの構築が必要になる。

flask はインタラクティブな flask shell なるものを搭載しているので、設定後はこちらで動作確認してみる。

ほんとはテストファーストで作っていくのが良いのだろうが、テストはメイントピックではないので順序的に後にする。

データベースは PostgreSQL を使う。 データモデルの作成には Flask-SQLAlchemy を、データベースのマイグレーションには Flask-Migrate を使う。

まず、requirements.txt に Flask-SQLAlchemy と Flask-Migrate を追加してコンテナを作り直す。 あと、PostgreSQL アダプターである psycopg2-binary も含める（コンパイル不要なバイナリ版である psycopg2-binary は開発やテスト環境では良いが、本番環境では psycopg2 の方が推奨されている）。 普通に pip install -r requirements.txt を実行しても良い。

Flask-Migrate>=4.0,<5.0
Flask-SQLAlchemy>=3.1,<4.0
psycopg2-binary>=2.9,<3.0

DB への接続

DB にデータを保存するデータモデルを作成していくが、その前に DB コンテナを追加する。 パスワードやユーザーを environment に指定し、あとで DBeaver などの DB クライアントでデータ確認ができるよう 5432 ポートを公開する。 また、volumes にはデータ同期するディレクトリを指定する。 ここでは main.py が存在するディレクトリ内に db/data ディレクトリを作成している。

version: "3.8"
services:
  app:
    build: .
    volumes:
      - .:/app
    ports:
      - 5001:5000
    command: bash -c "flask run --host=0.0.0.0"
    environment:
      - FLASK_APP=main
      - FLASK_DEBUG=on
      - TZ=Asia/Tokyo
  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

加えて、main.py で SQLAlchemy を利用した接続設定を行う。

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

app = Flask(__name__)

app.config["SQLALCHEMY_DATABASE_URI"] = 'postgresql://postgres:postgres@postgres/postgres'

db = SQLAlchemy()
db.init_app(app)

migrate = Migrate()
migrate.init_app(app, db)

# 以下省略

そうすると、docker compose up -d で PostgreSQL が起動します。

$ docker compose up -d
$ docker ps
CONTAINER ID   IMAGE                    COMMAND                  CREATED          STATUS          PORTS                    NAMES
e09f5794e54b   postgres:16.2-bookworm   "docker-entrypoint.s…"   40 seconds ago   Up 39 seconds   0.0.0.0:5432->5432/tcp   flask_web-postgres-1
a26ae3f560f2   flask_web-app            "bash -c 'flask run …"   40 seconds ago   Up 39 seconds   0.0.0.0:5001->5000/tcp   flask_web-app-1

DBeaver で接続すると空っぽの中身が覗けるでしょう。

データモデルの作成

モデルは main.py に書いていくが、後でファイルを分割する。 まず、事前に定義した db の Model を継承して Product と User を作成する。 Product は名前、価格、作成日時、更新日時を持つ。

# 省略
from datetime import datetime, timezone
from sqlalchemy.orm import Mapped, mapped_column

# 省略

class Product(db.Model):
    __tablename__ = 'products'
    id: Mapped[int] = mapped_column(db.Integer, primary_key=True)
    name: Mapped[str] = mapped_column(db.String(20), nullable=False)
    price: Mapped[int] = mapped_column(db.Integer, nullable=False)
    created: Mapped[datetime] = mapped_column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
    updated: Mapped[datetime] = mapped_column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
    def __repr__(self) -> str:
        return f'Product(id={self.id!r}, name={self.name!r}, price={self.price!r}, created={self.created!r}, updated={self.updated!r})'

class User(db.Model):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(db.Integer, primary_key=True)
    name: Mapped[str] = mapped_column(db.String(20), nullable=False)
    created: Mapped[datetime] = mapped_column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
    updated: Mapped[datetime] = mapped_column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
    def __repr__(self) -> str:
        return f'User(id={self.id!r}, name={self.name!r}, created={self.created!r}, updated={self.updated!r})'

DB マイグレーション

main.py にモデルを記述したら DB にテーブルを作成する。 Flask アプリを立ち上げるときに DB は作られているはずなので、あとは以下のコマンドを実行する。

# docker コンテナの中に入っておく
$ flask db migrate -m 'create products and users'
$ flask db upgrade

DBeaver で DB の中身を覗くと products テーブルと users テーブルが作られていることを確認できるかと思う。

Shell での確認

DB にテーブルができたので flask shell で SQLAlchemy の ORM を利用してデータの作成や取得を試してみる。

まずは Product の作成する。

$ flask shell
Python 3.12.2 (main, Mar 12 2024, 08:11:26) [GCC 12.2.0] on linux
App: main
Instance: /app/instance
>>> product = Product(name='Snack', price=100)
>>> db.session.add(product)
>>> db.session.commit()

Product を取得する。

>> product = db.session.execute(db.select(Product).filter_by(name='Snack')).scalar_one()
>> product
Product(id=1, name='Snack', price=100, created=datetime.datetime(2024, 3, 24, 18, 39, 26, 238131), updated=datetime.datetime(2024, 3, 24, 18, 39, 26, 238193))
>> products = db.session.query(Product).all()
>> products
[Product(id=1, name='Snack', price=100, created=datetime.datetime(2024, 3, 24, 18, 39, 26, 238131), updated=datetime.datetime(2024, 3, 24, 18, 39, 26, 238193))]

うまくいってそう。

EC サイトの画面を作る

DB のデータを用いて EC サイトの画面を作っていきます。 作るのは、商品一覧画面、商品詳細画面、商品購入履歴一覧画面、商品購入画面の4つ。

まず、ベースのテンプレートを作成しておく。 templates/base.html の中身は以下の通り。

<!-- base.html -->
<!DOCTYPE html>
<head>
  <title>{% block title %}{% endblock %}</title>
  <link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}" />
</head>
<body>
  <header>
    {% block header %}{% endblock %}
  </header>
  <main>
    {% block content %}{% endblock %}
  </main>
</body>

この block header とか block content の部分に個別のページの要素を作っていく。

商品一覧画面

base.html を拡張する形で templates/products.html を作成する。 このあと詳細画面を作るので詳細画面ページへの導線も作っておく。

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

{% block header %}
  <h1>{% block title %}Products{% endblock %}</h1>
{% endblock %}

{% block content %}
  <section>
    {% for product in products %}
      <dt>
        <a href="/products/{{ product.id }}">{{ product.name }}</a>
      </dt>
      <dd>{{ product.price }}</dd>
    {% endfor %}
  </section>
{% endblock %}

main.py の hello_world() メソッドを書いていたところの下あたりに以下を追加する。 ここでやってるのは DB の products テーブルからすべてのデータを取得し、それらを products 変数としてテンプレートに埋め込んでいる。

@app.route("/products")
def products():
    products = db.session.query(Product).all()
    return render_template('products.html', products=products)

上記にある通り、テンプレート側ではこの products 内の product を1つずつループしてその名前と価格を表示している。

商品詳細画面

詳細画面を一覧画面と別に作ることもできるが、ここでは一緒にまとめてみる。 以下のテンプレートは product が渡された場合は詳細画面として振る舞い、products が渡された場合には一覧画面として振る舞う。

<!-- products.html -->
{% block content %}
  <section>
    {% if product %}
      <dt>{{ product.name }}</dt>
      <dd>{{ product.price }}</dd>
    {% elif products %}
      {% for product in products %}
        <dt>
          <a href="/products/{{ product.id }}">{{ product.name }}</a>
        </dt>
        <dd>{{ product.price }}</dd>
      {% endfor %}
    {% endif %}
  </section>
{% endblock %}

main.py 側では product を返却するメソッドを作成する。

@app.route("/products/<product_id>")
def product(product_id):
    product = db.session.execute(db.select(Product).filter_by(id=product_id)).scalar_one()
    return render_template('products.html', product=product)

商品をカートに含める

カートの情報はセッションで記録しておく。 実際の運用では DB にカート内の情報を永続化したり色々工夫すれば良いと思う。 認証機能を含める場合は認証前はセッションで、認証後は DB に永続化とかできそう。 認証後もセッションで管理する方法もありうる。

なんでも良いが、add_cart() メソッドを以下のように追加する。 flask から session をインポートできるので、こいつに cart を持たせて指定された product_id を突っ込んでいく。

from flask import Flask, render_template, session, request, redirect, url_for

// 省略

@app.post('/add_cart')
def add_cart():
    cart_items = []
    if 'cart' in session:
        cart_items = session['cart']
    if request.form['product_id'] not in cart_items:
        cart_items.append(request.form['product_id'])
    session['cart'] = cart_items
    return redirect(url_for('cart'))

Flask のセッションについてはこちらを参照のこと。

で、商品の詳細画面にカートに加えるためのボタンを配置してあげる。 以下では in_cart という変数(Boolean)を確認してこれが false なら「カートに加える」ボタンが表示されるようにしている。 in_cart 変数は product() メソッドから返す。

<!-- products.html -->
<!-- 省略 -->
<dt>{{ product.name }}</dt>
<dd>{{ product.price }}</dd>
{% if not in_cart %}
  <form action="/add_cart" method="post">
    <input type="hidden" name="product_id" value="{{ product.id }}">
    <input type="submit" value="カートに加える">
  </form>
{% endif %}
<!-- 省略 -->

@app.route("/products/<product_id>")
def product(product_id):
    product = db.session.execute(db.select(Product).filter_by(id=product_id)).scalar_one()
    in_cart = False
    # session の cart に product_id が含まれる場合はここで True が返るようになっている
    if 'cart' in session and product_id in session['cart']:
        in_cart = True
    return render_template('products.html', product=product, in_cart=in_cart)

次はカート内の商品一覧画面だ。

<!-- cart.html -->
{% block content %}
  <section>
    {% for product in products %}
      <p>{{ product.name }}</p>
      <form action="/remove_cart" method="post">
        <input type="hidden" name="product_id" value="{{ product.id }}">
        <input type="submit" value="カートから取り除く">
      </form>
    {% endfor %}
    <a href="/products">商品を見る</a>
  </section>
{% endblock %}

API 側は以下の通り。 remove_cart() と cart() を追加している。

@app.post('/remove_cart')
def remove_cart():
    # session に cart が含まれる場合に限り、その中の product_id を削除するようにしている
    if 'cart' in session:
        cart_items = session['cart']
        cart_items.remove(request.form['product_id'])
        session['cart'] = cart_items
    return redirect(url_for('cart'))

@app.route("/cart")
def cart():
    # 空の products を初期化
    products = []
    if 'cart' in session:
        # cart に含まれる product_id から Product を DB から抽出して products に代入する
        cart_items = session['cart']
        app.logger.info(cart_items)
        products = db.session.query(Product).filter(Product.id.in_(cart_items)).all()
    # products を html 側でも使えるようにする
    return render_template('cart.html', products=products)

以上でカートへの追加と削除、および一覧表示ができるだろう。

商品の購入

商品の購入を実装する前に、ユーザーが商品を購入したという事実を記録するテーブルを作成する。 事前に users テーブルは作成していたので、products と users の中間テーブルである purchase_transactions テーブルを作成する。

class PurchaseTransaction(db.Model):
    __tablename__ = 'purchase_transactions'
    id: Mapped[int] = mapped_column(db.Integer, primary_key=True)
    product_id: Mapped[int] = mapped_column(db.ForeignKey('products.id'))
    user_id: Mapped[int] = mapped_column(db.ForeignKey('users.id'))
    product: Mapped['Product'] = db.relationship(back_populates='purchase_transactions')
    user: Mapped['User'] = db.relationship(back_populates='purchase_transactions')
    created: Mapped[datetime] = mapped_column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
    updated: Mapped[datetime] = mapped_column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
    def __repr__(self) -> str:
        return f'PurchaseTransaction(id={self.id!r}, product_id={self.product_id!r}, user_id={self.user_id!r} created={self.created!r}, updated={self.updated!r})'

合わせて、Product と User の方からも PurchaseTransaction を参照できるようにする。

class Product(db.Model):
    __tablename__ = 'products'
    id: Mapped[int] = mapped_column(db.Integer, primary_key=True)
    name: Mapped[str] = mapped_column(db.String(20), nullable=False)
    price: Mapped[int] = mapped_column(db.Integer, nullable=False)
    purchase_transactions: Mapped[List['PurchaseTransaction']] = db.relationship(back_populates='product', cascade='all, delete-orphan') # product.purchase_transactions というふうに参照することができる
    created: Mapped[datetime] = mapped_column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
    updated: Mapped[datetime] = mapped_column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
    def __repr__(self) -> str:
        return f'Product(id={self.id!r}, name={self.name!r}, price={self.price!r}, created={self.created!r}, updated={self.updated!r})'

class User(db.Model):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(db.Integer, primary_key=True)
    name: Mapped[str] = mapped_column(db.String(20), nullable=False)
    purchase_transactions: Mapped[List['PurchaseTransaction']] = db.relationship(back_populates='user', cascade='all, delete-orphan') # user.purchase_transactions というふうに参照することができる
    created: Mapped[datetime] = mapped_column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
    updated: Mapped[datetime] = mapped_column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
    def __repr__(self) -> str:
        return f'User(id={self.id!r}, name={self.name!r}, created={self.created!r}, updated={self.updated!r})'

一応 flask shell で作成テストをしてみる。

root@d4d07ffc9c48:/app# flask shell
Python 3.12.2 (main, Mar 12 2024, 08:11:26) [GCC 12.2.0] on linux
App: main
Instance: /app/instance
>>>
>>> user = db.session.execute(db.select(User).filter_by(id=1)).scalar_one()
>>> product = db.session.execute(db.select(Product).filter_by(id=1)).scalar_one()
>>> purchase_transaction = PurchaseTransaction(user=user, product=product)
>>> db.session.add(purchase_transaction)
>>> db.session.commit()
>>>
>>> user.purchase_transactions
[PurchaseTransaction(id=1, product_id=1, user_id=1 created=datetime.datetime(2024, 1, 1, 12, 0, 0, 0), updated=datetime.datetime(2024, 1, 1, 12, 0, 0, 0))]
>>>
>>> product.purchase_transactions
[PurchaseTransaction(id=1, product_id=1, user_id=1 created=datetime.datetime(2024, 1, 1, 12, 0, 0, 0), updated=datetime.datetime(2024, 1, 1, 12, 0, 0, 0))]

関連はうまくいってそうな感じ。 購入時に呼び出す API は以下の purchase() となる。購入者の user はハードコーディングで id=1 を指定しているが、実際は session などでユーザー情報を保持してそこから id を取り出すような形になりそう。 購入後は success.html を表示する。 実際の EC サイトでは Stripe などに購入リクエストを投げる API を PurchaseTransaction 作成の前後に挟むことになるかと思う。

@app.post("/purchase/<product_id>")
def purchase(product_id):
    user = db.session.execute(db.select(User).filter_by(id=2)).scalar_one()
    product = db.session.execute(db.select(Product).filter_by(id=product_id)).scalar_one()
    if user and product:
        purchase_transaction = PurchaseTransaction(user=user, product=product)
        db.session.add(purchase_transaction)
        db.session.commit()
    return render_template('success.html', product=product)

{% block content %}
  <section>
    <p>{{ product.name }}の購入に成功しました！</p>
    <a href="/products">商品を見る</a>
    <a href="/cart">カートを見る</a>
  </section>
{% endblock %}

一般的に EC サイトではカートに入れた商品を購入することになると思うので、上記 API を呼び出す購入ボタンはカートの画面に配置してみる。 今回は購入は商品ごとに行えるようにする。一括で複数まとめて購入する機能は作らない。

カート画面はこんな感じ。

<!-- cart.html -->
{% block content %}
  <section>
    {% for product in products %}
      <p>{{ product.name }}</p>
      <form action="/purchase/{{ product.id }}" method="post">
        <input type="hidden" name="product_id" value="{{ product.id }}">
        <input type="submit" value="購入する">
      </form>
      <form action="/remove_cart" method="post">
        <input type="hidden" name="product_id" value="{{ product.id }}">
        <input type="submit" value="カートから取り除く">
      </form>
    {% endfor %}
    <a href="/products">商品を見る</a>
  </section>
{% endblock %}

以上でカートからの購入と購入後の成功画面表示ができるようになった。 DBeaver などで DB の中を覗くと purchase_transactions のレコードが作成されていることを確認できるだろう。

商品の購入履歴

最後に、購入した商品の一覧を表示してみる。 transactions() を追加し、transaction.html を表示できるようにしておく。

@app.route("/transactions")
def transactions():
    purchase_transactions = db.session.query(PurchaseTransaction).all()
    return render_template('transactions.html', transactions=purchase_transactions)

<!-- transactions.html -->
{% block content %}
  <section>
    {% for transaction in transactions %}
      <dt>
        {{ transaction.product.name }}を購入しました。
      </dt>
      <dd>{{ transaction.product.price }}円支払いました。</dd>
    {% endfor %}
    <a href="/products">商品を見る</a>
    <a href="/cart">カートを見る</a>
  </section>
{% endblock %}

あとは、cart.html に購入履歴画面への動線を設けておく。

<!-- cart.html -->
<!-- 省略 -->
<a href="/transactions">購入履歴を見る</a>
<!-- 省略 -->

前項で追加した「購入する」ボタンをクリックすると、購入した商品が transactions.html に表示されるのを確認できるだろう。

を仕舞い

ここではフォームについては触れなかったが、html のフォームをバリデーション込みでいい感じにしてくれるライブラリに Flask WTF がある。

WTForm というライブラリを flask 向けに手を加えたライブラリなので、本家のドキュメントも参考になるかも。

これを使えば CSRF の対策を行うこともできるので、次回の Flask 記事で触れることにする。 CSRF の対策としては Origin ヘッダーを使う方法も最近流行ってるっぽいことを聞いたので、その辺も触れるつもり。

あと、ここで書いたソースコードはこちらにまとめてます。