概要
ポイントは SQLAlchemy の db オブジェクトを管理するモジュールを別にする点です
そこでモデルなども管理することで循環参照させないようにします
環境
- macOS 10.15.5
- MySQL 8.0.19
- Python 3.8.3
- Flask-SQLAlchemy 2.4.3
ライブラリインストール
今回はマイグレートにも対応するので Flask-Migrate もインストールします
pipenv install Flask-Marshmallow Flask-SQLAlchemy Flask-Migrate mysqlclient marshmallow-sqlalchemy
データベースを管理するモジュールの作成
今回の肝になるモジュールです
ポイントはここではオブジェクトの作成とモデルの管理を行うだけで Flask のコンテキストの埋め込みは行いません
それは Flask アプリを管理するモジュールで行う点です
そうすることで実際に SQL を発行すると処理を別モジュールとして更に管理できるようになります
vim my_app/database.py
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
db = SQLAlchemy()
ma = Marshmallow()
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(128))
age = db.Column(db.Integer)
def __repr__(self):
return "%s,%s,%i" % (self.id, self.name, self.age)
class UserSchema(ma.Schema):
class Meta:
fields = ("id", "name", "age")ライブラリ
例えばモデルをもとに実際に CRUD を行う場合は別モジュールを作成します
参照先が Flask アプリでなく切り離した database.py だけになります
vim my_app/lib/__init__.py
from my_app.database import db, User, UserSchema
class UserCRUD():
def add(self, name, age):
user = User(name=name, age=age)
db.session.add(user)
db.session.commit()
return 'ok'
def select(self):
return UserSchema(many=True).dump(User.query.all())
def _select_one(self, id):
return User.query.get(id)
def delete(self, id):
db.session.delete(self._select_one(id))
db.session.commit()
return 'ok'
def update(self, id, name, age):
user = self._select_one(id)
user.name = name
user.aget = age
db.session.add(user)
db.session.commit()
return 'ok'Flask アプリ
最後に Flask アプリを管理するモジュールを作成します
database.py に定義した SQLAlchemy のオブジェクトと Marshmallow のオブジェクトはここで app を使って Flask 上で扱えるようにします
少し気持ち悪いですがマイグレートの定義はここに記載しています (もしかするとこれも別にする方法がありそうですが)
from flask import Flask
from flask_migrate import Migrate
from my_app.database import db, ma
from my_app.lib import UserCRUD
app = Flask(__name__)
app.debug = True
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql+mysqldb://root:@localhost/test?charset=utf8"
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
db.init_app(app)
migrate = Migrate(app, db)
ma.init_app(app)
@app.route('/add')
def add():
crud = UserCRUD()
crud.add("hawksnowlog", 10)
return 'ok'マイグレートする
今回の構成であればマイグレートにも対応しています
FLASK_APP=my_app pipenv run flask db initFLASK_APP=my_app pipenv run flask db migrate -m "Initial migrate"FLASK_APP=my_app pipenv run flask db upgrade"
動作確認
マイグレートしてテーブルができたら動作確認してみましょう
FLASK_APP=my_app pipenv run lask runcurl 'localhost:5000/add
ちゃんとデータを挿入されているのが確認できると思います
最後に
Flask-SQLAlchemy で循環参照しないコツを紹介しました
公式のドキュメントのクイックスタートは簡単に書けるようになっていますがそのまま 1 つのモジュールで進めると大変なことになるので注意しましょう
おまけ: テストを書くには
テストを書く場合はテスト側でデモ用の app を作成し登録する必要があります
pytest の場合はこんな感じです
from flask import Flask
from my_app.database import db
from my_app.lib import UserCRUD
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql+mysqldb://root:@localhost/test?charset=utf8"
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
db.init_app(app)
app.app_context().push()
def test_add():
crud = UserCRUD()
ret = crud.add("hawksnowlog_test", 99)
assert ret == 'ok'PYTHONPATH=./ pipenv run pytest test
上記だと普通にデータベースが起動していることが前提でかつ実際にデータも入ってしまうので mock する場合は monkeypatch を db.session.commit に対して当てれば OK です
from flask import Flask
from my_app.database import db
from my_app.lib import UserCRUD
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql+mysqldb://root:@localhost/test?charset=utf8"
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
db.init_app(app)
app.app_context().push()
def test_add(monkeypatch):
monkeypatch.setattr(db.session, "commit", lambda: None)
crud = UserCRUD()
ret = crud.add("hawksnowlog_test", 99)
assert ret == 'ok'
0 件のコメント:
コメントを投稿