前回まででFlaskの基本的な使い方を学べたものとして、もう少し実践的なアプリケーションを実装します。

新たに空のプロジェクトを作成します。

前提条件

DBはMySQLを使用するものとし、既に以下の設定を行っているものとします。

127.0.0.1:3306 で接続可能な状態
スキーマmy_schemaを作成済み
ユーザーはroot、パスワードもroot
テーブルには何もない状態

PythonでどうDBを扱うのか

いきなり話が逸れますが、ORMという技術を使わない実装方法について説明します。

まずはDBに対応したDB Driverを探しましょう。

DB	Driver
MySQL	PyMySQL
PostgreSQL	Psycopg
SQLServer	pyodbc
SQLite	sqlite3

DBに合わせたDriverを使い、SQLを実行します。
以下はMySQLとSQLServerの比較です。

MySQLの場合

import pymysql.cursors

# Connect to the database
connection = pymysql.connect(host='localhost',
                             user='user',
                             password='passwd',
                             database='db',
                             cursorclass=pymysql.cursors.DictCursor)

with connection:
    with connection.cursor() as cursor:
        # Create a new record
        sql = "INSERT INTO `users` (`email`, `password`) VALUES (%s, %s)"
        cursor.execute(sql, ('webmaster@python.org', 'very-secret'))

    # connection is not autocommit by default. So you must commit to save
    # your changes.
    connection.commit()

SQLServerの場合

import pyodbc 
# Some other example server values are
# server = 'localhost\sqlexpress' # for a named instance
# server = 'myserver,port' # to specify an alternate port
server = 'tcp:myserver.database.windows.net' 
database = 'mydb' 
username = 'myusername' 
password = 'mypassword' 
cnxn = pyodbc.connect('DRIVER={ODBC Driver 17 for SQL Server};SERVER='+server+';DATABASE='+database+';UID='+username+';PWD='+ password)
cursor = cnxn.cursor()

#Sample insert query
count = cursor.execute("""
INSERT INTO SalesLT.Product (Name, ProductNumber, StandardCost, ListPrice, SellStartDate) 
VALUES (?,?,?,?,?)""",
'SQL Server Express New 20', 'SQLEXPRESS New 20', 0, 0, CURRENT_TIMESTAMP).rowcount
cnxn.commit()

DBに接続してINSERTするだけのコードですがかなり実装に差があるのがわかると思います。
特定のDBしか使用しないならいいでしょうが、そうとは限らないでしょう。

「最初はシンプルにSQLiteで実装して、後から別のDBに変えよう」
「SQLServerで実装したが、Json形式のサポートが弱いからPostgreSQLに変えよう」

ということはままあります。

今回はそういった煩わしさをなくしたいので、ORM、Migrationといった技術を使い、SQLを書くのではなくDBをインスタンスのように扱います。

パッケージインストール

pip install Flask Flask-Migrate flask-restx Flask-SQLAlchemy mysqlclient

Flask-SQLAlchemyがORM、Flask-MigrateがMigrationのためのパッケージです。

SQLAlchemy というパッケージもあります。Flask-SQLAlchemyはFlask拡張パッケージとして実装されており、FlaskでORMを利用する際に便利です。

app.py の作成

まずはメインファイルを作成しましょう。

app.py

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

app = Flask(__name__)

# flask-restxの設定
app.config['RESTX_MASK_SWAGGER'] = False
api = Api(app, version='1.0', title='Sample API', description='A sample API')

# Flask-SQLAlchemyの設定
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:root@127.0.0.1/my_schema?charset=utf8mb4'
db = SQLAlchemy(app)

# Flask-Migrateの設定
migrate = Migrate(app, db)

# 別ファイルをインポートする(後で作成)
from routes.user import user_module
app.register_blueprint(user_module)

Userモデルの作成

models/user

from datetime import datetime

from app import db, api
from sqlalchemy import Column, Integer, String, Boolean, Date, DateTime, Float, Text
from flask_restx import fields


# DB固有の型はこう使う
# from sqlalchemy.dialects.mysql import TINYINT, TIMESTAMP


class User(db.Model):
    __tablename__ = 'user'
    id = Column(Integer, index=True, primary_key=True, autoincrement=True)
    username = Column(String(80), nullable=False)
    email = Column(String(40), nullable=False)


UserDto = api.model('UserDto', {
    'id': fields.Integer(min=0, description='primary_key'),
    'username': fields.String(min_length=0, max_length=80),
    'email': fields.String(min_length=0, max_length=40),
})

UserInsertDto = api.model('UserInsertDto', {
    'username': fields.String(required=True, min_length=0, max_length=80),
    'email': fields.String(required=True, min_length=0, max_length=40),
})

UserUpdateDto = api.model('UserUpdateDto', {
    'username': fields.String(min_length=0, max_length=80),
    'email': fields.String(min_length=0, max_length=40),
})

UserクラスがMySQL上のuserテーブルと対応します。

OpenAPI定義のため以下の３つのDtoも定義してあります。

UserDto: Select時のレスポンス
UserInsertDto: Insert時のリクエスト
UserUpdateDto :Update時のリクエスト

Userルートの定義

routes/user.py

from flask import request, Blueprint
from flask_restx import Resource
from models.user import User, UserDto, UserInsertDto, UserUpdateDto
from app import db, api

user_module = Blueprint('user_module', __name__)


@api.route('/user')
class Route(Resource):
    @api.marshal_list_with(UserDto)
    def get(self):
        """ 全レコードを取得します """
        users = db.session.query(User).all()
        return users

    @api.expect(UserInsertDto)
    @api.marshal_with(UserDto)
    def post(self):
        """ レコードを１件追加します """
        json_body = request.json

        new_user = User(**json_body)
        db.session.add(new_user)
        db.session.commit()

        return new_user


@api.route('/user/<int:id>')
class RouteWithId(Resource):
    @api.marshal_with(UserDto)
    def get(self, id: int):
        """ IDを指定してレコードを１件取得します """
        user = db.session.query(User).filter(User.id == id).one()
        return user

    @api.expect(UserUpdateDto)
    @api.marshal_with(UserDto)
    def patch(self, id: int):
        """ IDを指定してレコードを１件更新します """
        json_body = request.json
        update_user = db.session.query(User).filter(User.id == id).one()
        for k, v in json_body.items():
            setattr(update_user, k, v)
        db.session.commit()
        return update_user

    @api.marshal_with(UserDto)
    def delete(self, id: int):
        """ IDを指定してレコードを１件削除します """
        delete_user = db.session.query(User).filter(User.id == id).one()
        db.session.delete(delete_user)
        db.session.commit()
        return delete_user

userテーブルに対するCRUDを行えるようGET(Select), POST(Insert), PATCH(Update), DELETE(Delete)を定義しています。

GET /userで全レコード取得、GET /user/{id} で１件取得します。
複数レコード取得する場合はCRUD + L(LはList)というらしいです。

@api.expect(UserInsertDto)、@api.marshal_with(UserDto)の部分が重要で

@api.expect(UserInsertDto)がサーバへ送られてくるデータの形の指定。
@api.marshal_with(UserDto)がサーバが返すべきデータの形の指定だけではなく、Userクラスからjson形式への変換も行ってくれます。

試しに @api.marshal_with(UserDto)を削除すると正しくレスポンスを返せません。
-> TypeError: Object of type User is not JSON serializableになります

また、@api.marshal_list_with(UserDto)はUserクラスを複数(配列で)返す場合の書き方で、@api.marshal_with(UserDto, as_list=True)と書いても同じです。

DBの作成

実装はできましたがDBはまだ作成されていません。作成しましょう。

マイグレーションリポジトリの作成

初回のみ必要。やり直す場合はmigrationsフォルダを削除してから行う。

flask db init

PS D:\Python\flask_qiita> flask db init
Creating directory D:\Python\flask_qiita\migrations ...  done
Creating directory D:\Python\flask_qiita\migrations\versions ...  done
Generating D:\Python\flask_qiita\migrations\alembic.ini ...  done
Generating D:\Python\flask_qiita\migrations\env.py ...  done
Generating D:\Python\flask_qiita\migrations\README ...  done
Generating D:\Python\flask_qiita\migrations\script.py.mako ...  done
Please edit configuration/connection/logging settings in 'D:\\Python\\flask_qiita\\migrations\\alembic.ini' before proceeding.

migrationsフォルダが作成されます。この時点では初期設定のみです。

新規作成後はmigrations/env.pyのcontext.configure()のkwargsに
compare_type=Trueとcompare_server_default=Trueを追加した方がいいかもしれません。
追加しないと「型」と「デフォルト値」の変更が検知されません。

def run_migrations_online():
    # ...
    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,            # <- 追加
            compare_server_default=True,  # <- 追加
            process_revision_directives=process_revision_directives,
            **current_app.extensions['migrate'].configure_args
        )

マイグレーションファイルの作成

flask db migrate

PS D:\Python\flask_qiita> flask db migrate
INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.autogenerate.compare] Detected added table 'user'
INFO  [alembic.autogenerate.compare] Detected added index 'ix_user_id' on '['id']'
Generating D:\Python\flask_qiita\migrations\versions\5ebb528e8a5e_.py ...  done

現在のDBの状態とFlaskのModelとの差を検出できました。
この時点ではalenmbic_versionというテーブルが作成されたのみで、userテーブルは作成されていません。

alenmbic_versionテーブルはDBのバージョン管理のためのテーブルです。

マイグレーションの適用

flask db upgrade

PS D:\Python\flask_qiita> flask db upgrade             
INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> 5ebb528e8a5e, empty message

テーブル作成ができました。flask runでサーバを起動すればSwaggerUIからDBの更新が行えます。

DB移行を行う

DBへの列の追加を試みます。
Userクラスの変更だけでDBの移行自体は行えますがOpenAPIとの整合性を維持するため ...Dtoも一緒に変更します。

models/user.py

from datetime import datetime

from app import db, api
from sqlalchemy import Column, Integer, String, Boolean, Date, DateTime, Float, Text
from flask_restx import fields


# DB固有の型はこう使う
# from sqlalchemy.dialects.mysql import TINYINT, TIMESTAMP


class User(db.Model):
    __tablename__ = 'user'
    id = Column(Integer, index=True, primary_key=True, autoincrement=True)
    username = Column(String(80), nullable=False)
    email = Column(String(40), nullable=False)
    birth_day = Column(Date, nullable=False, default=datetime.now)
    height = Column(Float, nullable=False, default=0.0)
    memo = Column(Text, nullable=False, default='')
    created_at = Column(DateTime(timezone=True), nullable=False, default=datetime.now)
    updated_at = Column(DateTime(timezone=True), nullable=False, default=datetime.now, onupdate=datetime.now)
    is_active = Column(Boolean, nullable=False, default=True)


UserDto = api.model('UserDto', {
    'id': fields.Integer(min=0, description='primary_key'),
    'username': fields.String(min_length=0, max_length=80),
    'email': fields.String(min_length=0, max_length=40),
    'birth_day': fields.Date(example=datetime.now().strftime('%Y-%m-%d')),
    'height': fields.Float(example=0.0),
    'memo': fields.String(example='memo'),
    'created_at': fields.DateTime(example=datetime.now().strftime('%Y-%m-%dT%H:%M:%S')),
    'updated_at': fields.DateTime(example=datetime.now().strftime('%Y-%m-%dT%H:%M:%S')),
    'is_active': fields.Boolean,
})

UserInsertDto = api.model('UserInsertDto', {
    'username': fields.String(required=True, min_length=0, max_length=80),
    'email': fields.String(required=True, min_length=0, max_length=40),
    'birth_day': fields.Date(default=datetime.now().strftime('%Y-%m-%d')),
    'height': fields.Float(default=0.0),
    'memo': fields.String(default=''),
    'is_active': fields.Boolean(default=True),
})

UserUpdateDto = api.model('UserUpdateDto', {
    'username': fields.String(min_length=0, max_length=80),
    'email': fields.String(min_length=0, max_length=40),
    'birth_day': fields.Date(),
    'height': fields.Float(),
    'memo': fields.String(),
    'is_active': fields.Boolean(),
})

マイグレーションファイルの作成 & 適用

flask db migrate
flask db upgrade

PS D:\Python\flask_qiita> flask db migrate
Creating directory D:\Python\flask_qiita\migrations ...  done
Creating directory D:\Python\flask_qiita\migrations\versions ...  done
Generating D:\Python\flask_qiita\migrations\alembic.ini ...  done
Generating D:\Python\flask_qiita\migrations\env.py ...  done
Generating D:\Python\flask_qiita\migrations\README ...  done
INFO  [alembic.autogenerate.compare] Detected added column 'user.memo'
INFO  [alembic.autogenerate.compare] Detected added column 'user.created_at'
INFO  [alembic.autogenerate.compare] Detected added column 'user.updated_at'
INFO  [alembic.autogenerate.compare] Detected added column 'user.is_active'
Generating D:\Python\flask_qiita\migrations\versions\afc20f79324a_.py ...  done

PS D:\Python\flask_qiita> flask db upgrade
INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade 5ebb528e8a5e -> afc20f79324a, empty message

Modelの変更と同じようにDBが変更され、移行できました。

補足など

ORMを使えばDBの扱いが楽になりますし、Migrationがあれば多少の変更なら問題なしです。
とはいえMigrationによるDB移行は万能ではありません。大きな変更はうまく移行できない場合もありますので過信しすぎないようにしてください。