前回までで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移行は万能ではありません。大きな変更はうまく移行できない場合もありますので過信しすぎないようにしてください。