3. ORM + MigrationでSQLを一切書かずにDBのCRUDや移行を行う

前回までで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の比較です。

```python: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()
```

```python: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](https://ja.wikipedia.org/wiki/%E3%82%AA%E3%83%96%E3%82%B8%E3%82%A7%E3%82%AF%E3%83%88%E9%96%A2%E4%BF%82%E3%83%9E%E3%83%83%E3%83%94%E3%83%B3%E3%82%B0)、Migrationといった技術を使い、SQLを書くのではなくDBをインスタンスのように扱います。

---


# パッケージインストール

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

Flask-SQLAlchemyが[ORM](https://ja.wikipedia.org/wiki/%E3%82%AA%E3%83%96%E3%82%B8%E3%82%A7%E3%82%AF%E3%83%88%E9%96%A2%E4%BF%82%E3%83%9E%E3%83%83%E3%83%94%E3%83%B3%E3%82%B0)、Flask-MigrateがMigrationのためのパッケージです。

> [SQLAlchemy](https://docs.sqlalchemy.org/en/14/) というパッケージもあります。[Flask-SQLAlchemy](https://flask-sqlalchemy.palletsprojects.com/en/2.x/)はFlask拡張パッケージとして実装されており、FlaskでORMを利用する際に便利です。

# app.py の作成

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

```python: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モデルの作成

```python: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ルートの定義


```python: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の作成

![キャプチャ.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/611565/99bde646-618c-2e1c-fa6b-22fd1391b615.png)


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

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

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


```bash
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`を追加した方がいいかもしれません。
> 追加しないと「型」と「デフォルト値」の変更が検知されません。

  ```python
  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
          )
  ```


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

```bash
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のバージョン管理のためのテーブルです。

##### マイグレーションの適用

```bash
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
```

![キャプチャ2.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/611565/d611125a-498c-12a8-a677-b3aa7f3d0283.png)

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

# DB移行を行う

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

```python: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(),
})

```

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


```bash
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
```

![キャプチャ3.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/611565/156e4301-c1aa-afd6-814b-8137b96a3f13.png)

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

---

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



- [1. Flask 環境構築、簡単なバックエンドサーバー実装](https://qiita.com/YamaguchiRyuta/items/69ab10959767e27b8787)
- [2. FlaskでOpenAPI定義の自動出力、Swagger UIページの作成](https://qiita.com/YamaguchiRyuta/items/617c0299ea7277af4d9b)
- [3. FlaskでORM + MigrationでSQLを一切書かずにDBのCRUDや移行を行う](https://qiita.com/YamaguchiRyuta/items/dd88089b2fd41ea91745) <-
- [4. FlaskアプリをAWSにデプロイ](https://qiita.com/YamaguchiRyuta/items/ffc64586a40dd9a76742)