タイトルの通り
Flask はデフォルトだと 500 エラーが発生したときに HTML 用のページを返却します
app = Flask(__name__)
@app.errorhandler(500)
def not_found(error):
return Response(
json.dumps({'error': 'internal error'}),
status=500
)アプリを起動する際のオプションで debug オプションというのがありこれが True だとデフォルトの HTML を表示し続けてしまうので False にします
app.run(debug=False)これを結構忘れがちなので注意です
Ruby だと puts や pp などでオブジェクト内の attributes の参照ができます
もしくは instance_variables をコールしても attributes の参照ができます
メソッドの参照をするには methods などを使えばできます
Python3 でも方法があるようなのでやってみました
__dict__ を使う対象のオブジェクトの __dict__ を参照します
これで attributes が key - value 形式で取得できます
class Hoge(object):
def __init__(self):
self.a = 'a'
self.b = 'b'
def func1(self):
return self.a
h = Hoge()
print(h.__dict__)ただし以下のように別のクラスのオブジェクトを持っている場合は展開しません
class Fuga(object):
def __init__(self):
self.c = 'c'
self.d = 'd'
def func1(self):
return self.c
class Hoge(object):
def __init__(self):
self.a = 'a'
self.b = 'b'
self.f = Fuga()
def func1(self):
return self.a
h = Hoge()
print(h.__dict__){'a': 'a', 'b': 'b', 'f': <__main__.Fuga object at 0x10e34aa20>}中身は __dict__ と同じらしいです
最後の部分を以下のように書き換えれば OK です
print(vars(h))ただしこの場合もネスト的に展開はしてくれません
いきなり複雑になりますが以下の通りです
dir() と getattr() を使います
class Fuga(object):
def __init__(self):
self.c = 'c'
self.d = 'd'
def func1(self):
return self.c
class Hoge(object):
def __init__(self):
self.a = 'a'
self.b = 'b'
self.f = Fuga()
def func1(self):
return self.a
h = Hoge()
methods = [method for method in dir(h) if callable(getattr(h, method))]
print(methods)Python3 でオブジェクトが持つ attributes とメソッドを参照する方法を紹介しました
正直 Ruby のように簡単にはいきませんができるのはできました
SQLAlchemy を使う際にいろいろなコンポーネントが必要になります
多くのサンプルではそれぞれを 1 つファイルで扱っているようですが、あまりキレイな書き方ではないかなと個人的には思います
なので、各役割ごとにクラスを作成し管理するようにしてみました
提案的な記事なので参考程度に見ていただけると嬉しいです
DB サーバに接続する engine を管理します
SQLAlchemy の場合 engine は create_engine で作成します
DB サーバに接続する必要があるので DB サーバのへの接続情報 (ユーザ名、パスワード、ホスト名、データベース名など) はこのクラスで管理します
from sqlalchemy import create_engine
class BaseEngine(object):
def __init__(self):
username = 'username'
password = 'password'
hostname = 'localhost'
dbname = 'db_server'
url = 'mysql+mysqldb://{}:{}@{}/{}?charset=utf8'.format(username, password, hostname, dbname)
self.engine = create_engine(url, echo=True)DB サーバに接続するための情報はサンプルなので決めうちにしていますが環境変数や設定ファイルから持ってくるようにしても良いと思います
キーワード引数で受け取れるようにしても良いと思います
Session は DB への CRUD 操作するために必須のオブジェクトになります
Session は engine を元に生成する必要があるため先ほど作成した BaseEngine クラスを継承します
今回は先ほど作成したファイル (base_engine.py) で管理します (特に理由はないので分けても OK です)
ただし、クラスは Session を管理するためのクラスを作成します
from sqlalchemy.orm import sessionmaker
class BaseSession(BaseEngine):
def __init__(self):
super().__init__()
Session = sessionmaker(bind=self.engine)
self.session = Session()実際にアプリケーションから CRUD 操作をしたい場合はこのクラスを使います
使い方は後ほど紹介します
テーブルの構成を管理するクラスです
SQLAlchemy で言うところの Base を継承して作成するクラスになります
Column などを使ってテーブルのスキーマを定義します
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String(20))
age = Column(Integer)このクラスで肝になるのは Base です
この Base クラスは DB に対してテーブルを作成する場合にも使います
いろいろなところで Base を declarative_base するのは気持ち悪いので Base はここだけで扱うようにします
作成したスキーマ情報を元に DB サーバに対してテーブルを作成したいと思います
DB サーバに対してテーブルを作成する場合には Base と engine が主に必要になります
それらは専用で管理しているクラスとしてすでに作成済みなのでそれらを import して使います
from base import BaseEngine
from models import User, Base
class Migration(object):
def __init__(self):
self.e = BaseEngine().engine
def users(self):
Base.metadata.create_all(self.e)
if __name__ == '__main__':
Migration().users()エンジンを作成して、そのエンジンを元に create_all をコールするだけです
これで BaseEngine で指定したデータベース配下にテーブルを作成することができます
基本的にマイグレートするためのクラスは一度しか実行しない想定です
(SQLAlchemy の場合、すでにテーブルが存在する場合は何もしないので何度実行しても問題はないです)
作成したクラスを元に実際に SELECT 文を発行するクラスを作成してみます
必要になるのは Session を管理しているクラスと操作したいテーブルを管理しているモデルのクラスになります
from base import BaseSession
from models import User
class Users(BaseSession):
def __init__(self):
super().__init__()
def select(self):
for i in self.session.query(User).order_by(User.id):
print(i.name)
if __name__ == '__main__':
cli = Users()
cli.select()サンプルはただ print しているだけです
このメソッドの返り値を User の配列などにしてもいいかなと思います
今回の構成の場合、こんな感じで CRUD したいテーブルごとにクラスを作成できるので、コードの管理がわかりやすくなるかなと思っています
必要に応じて更に CRUD 用のベースクラスなども作成できるかなと思います
SQLAlchemy を使う場合にどのようなモジュール、クラスの構成にしたほうが良いか考えてみました
もしかしたら探せばベストプラクティスが出てくるかもしれません
今回紹介したのは最低限の役割のみになります
トランザクション処理や Alter 処理、外部キー制約が絡んでくる場合にはもう少し考慮することが増えますが基本的には今回の構成を応用するだけかなと思っています
SQLAlchemy を使う人の参考になれば幸いです
flasgger で作成したアプリを unittest でテストしてみました
アプリは Marshmallow Schemas で作成しています
Mock との連携方法も紹介します
# coding: utf-8
from flask import Flask, jsonify
from flasgger import Schema, Swagger, SwaggerView, fields
class CategorySchema(Schema):
id = fields.Int()
name = fields.Str(required=True)
class PetSchema(Schema):
category = fields.Nested(CategorySchema, many=True)
name = fields.Str()
class RandomView(SwaggerView):
summary = 'A cute furry animal endpoint.'
description = 'Get a random pet'
parameters = [
{
'name': 'id',
'in': 'path',
'required': True,
'type': 'integer'
}
]
responses = {
200: {
'description': 'A pet to be returned',
'schema': PetSchema
}
}
def get(self, id):
pet = self.external_api(id)
return jsonify(PetSchema().dump(pet).data)
def external_api(self, id):
return {'category': [{'id': id, 'name': 'rodent'}], 'name': 'Mickey'}
app = Flask(__name__)
app.add_url_rule(
'/random/<id>',
view_func=RandomView.as_view('random'),
methods=['GET']
)
if __name__ == '__main__':
app.run(debug=True)こんな感じのアプリです
/random/1 に GET すると適当な JSON が返ってきます
import unittest
import json
import test_app
class TestApp(unittest.TestCase):
def setUp(self):
self.app = test_app.app.test_client()
def testGet(self):
res = self.app.get('/random/1')
self.assertEqual(
{'category':[{'id':1,'name':'rodent'}],'name':'Mickey'},
res.get_json()
)
if __name__ == '__main__':
unittest.main()pipenv run python3 -m unittest test1.pyポイントは test_app.app.test_client() で Flask アプリ用のクライアントを作成している点です
このクライアントの get や post といったメソッドをコールすることでアプリにリクエストすることができます
次に unittest.MagicMock を使ってテストします
external_api というメソッドを Mock してみたいと思います
import unittest
from unittest.mock import patch, MagicMock
import json
import test10
class TestApp(unittest.TestCase):
def setUp(self):
self.app = test10.app.test_client()
@patch('test10.RandomView.external_api')
def testMockGet(self, mock):
mock.return_value = {'category':[{'id':100,'name':'rodent'}],'name':'Mickey'}
print(mock)
res = self.app.get('/random/1')
self.assertEqual(
{'category':[{'id':100,'name':'rodent'}],'name':'Mickey'},
res.get_json()
)
if __name__ == '__main__':
unittest.main()ポイントは @patch('test10.RandomView.external_api') です
関数まで patch 当てます
すると引数の mock に MagicMock クラスのオブジェクトが返ってきます
更に mock は patch のおかげで関数にもなっているのであとは return_value で返り値を設定するだけです
今回は id:100 を Mock が返すように設定しています
pipenv run python3 -m unittest test2.pyflasgger で作成したアプリに対して unittest を適用してみました
Mock も使ってみましたが SwaggerView で定義した内容を Mock することで意図するテストを書くことができました
公式にテストの方法は書いてませんでしたが、Flask の test_client と標準の unittest を使っているので特に追加のモジュールなしでテストできるのは嬉しい点だと思います
flasgger の 404 はデフォルトだと HTML が返ってきます
<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 3.2 Final//EN">
<title>404 Not Found</title>
<h1>Not Found</h1>
<p>The requested URL was not found on the server. If you entered the URL manually please check your spelling and try again.</p>API の場合は JSON のほうが良いという場合がほとんどだと思います
今回は flasgger で 404 ページをカスタマイズする方法を紹介します
flask の errorhandler という機能を使います
app = Flask(__name__)
@app.errorhandler(404)
def not_found(error):
return Response(
json.dumps({'error': 'not found path'}),
status=404
)これを適当な場所に定義すれば OK です
ちなみに必要な import は以下の通り
import json
from flask import Response
from flask import Flaskflasgger で 404 ページをカスタマイズする方法を紹介しました
flask の機能を使うことで解決することができます
flasgger は flask や marshmallow, apispec などいろいろなサードパティツールを使っているのでそれらを使うことで解決できることは多いと思います
flasgger には validation の機能がデフォルトで備わっています
今回は validation 機能を使って挙動を確認してみました
以下のサンプルアプリを元に挙動を確認します
POST のリクエストを受け取って処理する簡単なアプリです
# coding: utf-8
from flask import Flask, jsonify
from flasgger import Schema, Swagger, SwaggerView, fields
class CategorySchema(Schema):
id = fields.Int()
name = fields.Str(required=True)
class PetSchema(Schema):
category = fields.Nested(CategorySchema, many=True)
name = fields.Str(required=True)
class RandomView(SwaggerView):
parameters = [
{
'in': 'body',
'name': 'Pet',
'description': 'Register a pet',
'required': True,
'schema': PetSchema
}
]
responses = {
200: {
'description': 'Registered',
'schema': fields.Str()
}
}
def post(self):
return 'registered'
app = Flask(__name__)
app.add_url_rule(
'/random',
view_func=RandomView.as_view('random'),
methods=['POST']
)
Swagger(app)
if __name__ == '__main__':
app.run(debug=True)正常なリクエストは以下の通りです
validation 機能を入れることでこのリクエストがどうなるか確認します
curl -v -XPOST -H 'content-type: application/json' -d '{"category":[{"id":1,"name":"rodent"}],"name":"Mickey"}' 'localhost:5000/random'validation = True にするには View クラスのフィールドで有効にするだけです
class RandomView(SwaggerView):
parameters = [
{
'in': 'body',
'name': 'Pet',
'description': 'Register a pet',
'required': True,
'schema': PetSchema
}
]
responses = {
200: {
'description': 'Registered',
'schema': fields.Str()
}
}
validation = Trueこの設定でエラーとなる挙動は以下の通りです
content-type ヘッダがセットされていない場合リクエスト
curl -v -XPOST -d '{"category":[{"id":1,"name":"rodent"}],"name":"Mickey"}' 'localhost:5000/random'レスポンス
No data to validatecurl -v -XPOST -H 'content-type: application/json' -d '{"category":[{"id":1,"name":"rodent"}]}' 'localhost:5000/random''name' is a required property Failed validating 'required' in schema: で正しいスキーマ情報が表示されるこんな感じで最低限のチェックを行ってくれる感じです
現状はないようです
もしカスタマイズしたい場合は validation_function を指定して自分で validation 処理を実装する必要があるようです
がしかし、どうやら Marshmallow Schemas で validation_function を使用する方法はないです
Marshmallow Schemas の場合すべてクラスで定義します
validation_function を使うためには swagger の定義ファイル or 定義の dictionary オブジェクトが必要になります
無理矢理使うことはできなくはないですが、せっかくクラスで定義したものをファイル or dictionary でもう一度定義しなければいけないのはかなり微妙な感じになると思います
試していないですが marshmallow の機能で Schema からサンプルデータを突っ込んで JSON なり dictionary を生成する機能があるので、それを使えばできなくはないと思います
が、それも微妙かなと思います
Marshmallow schemas を使っている場合は以下のどちらかで validation するしかなさそうです
かなと思います
自分が調べた限りだとこの 2 つのどちらかになりそうです
Marshmallow Schemas で validation と validation_function 機能を試してみました
結論としてはデフォルトの validation を使わず自力で validation 用のメソッドを実装するのが一番良いかなと思います
情報が少ないので何とも言えませんがコードを直接見たりすれば他の解決方法が見つかるかもしれません (自分は探せませんでした、というか自力 validation を作る方が楽だと判断しました)
flasgger でパラメータに path を使っている場合に validation する方法を紹介します
flasgger にはおそらくデフォルトでは path をチェックする機能は備わっていないので自分で実装する必要があります
これをベースに validation 機能を付けてみます
path でパラメータを取るのでこれを validate してみます
# coding: utf-8
from flask import Flask, jsonify
from flasgger import Schema, Swagger, SwaggerView, fields
class CategorySchema(Schema):
id = fields.Int()
name = fields.Str(required=True)
class PetSchema(Schema):
category = fields.Nested(CategorySchema, many=True)
name = fields.Str()
class RandomView(SwaggerView):
summary = 'A cute furry animal endpoint.'
description = 'Get a random pet'
parameters = [
{
'name': 'id',
'in': 'path',
'required': True,
'type': 'integer'
}
]
responses = {
200: {
'description': 'A pet to be returned',
'schema': PetSchema
}
}
def get(self, id):
pet = {'category': [{'id': id, 'name': 'rodent'}], 'name': 'Mickey'}
return jsonify(PetSchema().dump(pet).data)
app = Flask(__name__)
app.add_url_rule(
'/random/<id>',
view_func=RandomView.as_view('random'),
methods=['GET']
)
if __name__ == '__main__':
app.run(debug=True)で起動して
curl localhost:5000/random/1で以下のようなレスポンスが返ってきます
{
"category": [
{
"id": 1,
"name": "rodent"
}
],
"name": "Mickey"
}validation 機能を入れる前は文字列でも問題ありません
これが数字以外の場合はエラーになるようにします
まず import 系を少し追加します
validation 時にエラーのレスポンスを直接返却する必要があるので、それに関するモジュールやクラスを import します
import json
from werkzeug.exceptions import abort
from flask import Response上記を追加しましょう
そして validation 用のメソッドを追加します
def my_validate(self, id):
try:
int(id)
except ValueError as e:
print(e)
abort(
Response(
json.dumps({'error': 'id must be set integer type', 'id': id}),
status=400
)
)path で取得した id 情報を validation 用のメソッドに渡します
そして integer に変換できるか調査して、もし Exception が発生したらエラーを返却します
あとはこのメソッドをコールするだけです
def get(self, id):
self.my_validate(id)
pet = {'category': [{'id': id, 'name': 'rodent'}], 'name': 'Mickey'}
return jsonify(PetSchema().dump(pet).data)修正後の全体のコードは以下の通りです
# coding: utf-8
import json
from werkzeug.exceptions import abort
from flask import Response
from flask import Flask, jsonify
from flasgger import Schema, Swagger, SwaggerView, fields
class CategorySchema(Schema):
id = fields.Int()
name = fields.Str(required=True)
class PetSchema(Schema):
category = fields.Nested(CategorySchema, many=True)
name = fields.Str()
class RandomView(SwaggerView):
summary = 'A cute furry animal endpoint.'
description = 'Get a random pet'
parameters = [
{
'name': 'id',
'in': 'path',
'required': True,
'type': 'integer'
}
]
responses = {
200: {
'description': 'A pet to be returned',
'schema': PetSchema
}
}
def my_validate(self, id):
try:
int(id)
except ValueError as e:
print(e)
abort(
Response(
json.dumps({'error': 'id must be set integer type', 'id': id}),
status=400
)
)
def get(self, id):
self.my_validate(id)
pet = {'category': [{'id': id, 'name': 'rodent'}], 'name': 'Mickey'}
return jsonify(PetSchema().dump(pet).data)
app = Flask(__name__)
app.add_url_rule(
'/random/<id>',
view_func=RandomView.as_view('random'),
methods=['GET']
)
if __name__ == '__main__':
app.run(debug=True)これで再度実行すると id の部分が文字列の場合にはエラーが返ってくるようになります
curl -v 'localhost:5000/random/a'{"error": "id must be set integer type", "id": "a"}実は flasgger には validation の機構が備わっています
validation と validation_function という機能があります
validation はデフォルトで用意された validator で有効/無効の値しか設定できません
True にした場合に有効になります
やってくれることは例えば
などです
今回のようにフォーマットチェックなどしたい場合には正直使えません
True にしても余計なことをチェックするケースが多いかなと思います
また validation_function に関してですがこれは基本的に POST 時のボディを validate するときに使うっぽいです (？)
今回のように path のフォーマットチェックをする場合は素直に validation 用のメソッドを作ってしまうほうが早いです
flasgger の Marshmallow Schemas で path パラメータをチェックするための独自の validation 機能を実装してみました
おそらくこの方法が一番てっとり早いと思います
途中で軽く触れた validation と validation_function の機能に関しても検証してみたいと思います
この辺りの情報はググっても出てこないので Github の issue やコードを見るしか方法がなさそうです
SQLArchemy は Python で使える ORM です
簡単な CRUD 操作からマイグレーションまで幅広い機能を提供しています
今回は簡単なテーブルの作成から CRUD 操作まで行ってみました
今回は MySQL にアクセスするので mysqlclient も合わせてインストールします
おそらく ModuleNotFoundError: No module named 'MySQLdb' のエラーが出ると思います
そして pipenv install MySQL-Python をインストールしようとしたのですがどうやらまだ Python3 に対応していないようです
なので PyMySQL をインストールしようとしたのですが状況は変わらずで最終的に mysqlclient をインストールすることで解決しました
mysql -u root -p -e "create database test;"test データベースを作成しておきましょう
from sqlalchemy import create_engine
url = 'mysql+mysqldb://user:password@localhost/test?charset=utf8'
engine = create_engine(url, echo=True)こんな感じです
ユーザ、パスワードの部分は適当に変更してください
test データベースに接続しています
test1.py にいろいろ追記していきます
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
url = 'mysql+mysqldb://root@localhost/test?charset=utf8'
engine = create_engine(url, echo=True)
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String(50))
fullname = Column(String(100))
password = Column(String(100))
def __repr__(self):
return "<User(name='%s', fullname='%s', password='%s')>" % (self.name, self.fullname, self.password)まず User クラスは Base クラスを継承します
そして __tablename__ 変数でテーブル名を指定します
あとは Column を使ってテーブルに定義するカラムを定義します
__repr__ は必須ではないですが実装しておくとレスポンスをキレイに見せることができます
先ほどのスキーマ定義から実際にテーブルを作成するところまで追加します
と言っても最後の 1 行を追加しているだけです (Base.metadata.create_all(engine))
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
url = 'mysql+mysqldb://root@localhost/test?charset=utf8'
engine = create_engine(url, echo=True)
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String(50))
fullname = Column(String(100))
password = Column(String(100))
def __repr__(self):
return "<User(name='%s', fullname='%s', password='%s')>" % (self.name, self.fullname, self.password)
Base.metadata.create_all(engine)これで test データベースは以下に users というテーブルが作成されています
mysql -u user -p password test -e "desc users;"+----------+--------------+------+-----+---------+----------------+
| Field | Type | Null | Key | Default | Extra |
+----------+--------------+------+-----+---------+----------------+
| id | int(11) | NO | PRI | NULL | auto_increment |
| name | varchar(50) | YES | | NULL | |
| fullname | varchar(100) | YES | | NULL | |
| password | varchar(100) | YES | | NULL | |
+----------+--------------+------+-----+---------+----------------+sqlalchemy.exc.CompileError: (in table 'users', column 'name'): VARCHAR requires a length on dialect mysql
というエラーになる場合はカラムを定義している String の部分で引数に文字数を指定しているか確認してください
VARCHAR はちゃんと文字数制限を入れないとエラーとなります
engine を元に Session を作成することでテーブルにアクセスすることができます
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
url = 'mysql+mysqldb://root@localhost/test?charset=utf8'
engine = create_engine(url, echo=True)
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String(50))
fullname = Column(String(100))
password = Column(String(100))
def __repr__(self):
return "<User(name='%s', fullname='%s', password='%s')>" % (self.name, self.fullname, self.password)
Base.metadata.create_all(engine)
u1= User(name='hawk', fullname='hawksnowlog', password='xxxxxxx')
Session = sessionmaker(bind=engine)
session = Session()
session.add(u1)
session.commit()最後の session.commit() するまでデータは挿入されません
なのでユーザを複数人登録したり更新処理なども行ってから commit することができます
いわゆるトランザクション処理になります
ちなみに commit する前の状態にロールバックしたい場合は session.rollback() を呼び出せば OK です
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
url = 'mysql+mysqldb://root@localhost/test?charset=utf8'
engine = create_engine(url, echo=True)
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String(50))
fullname = Column(String(100))
password = Column(String(100))
def __repr__(self):
return "<User(name='%s', fullname='%s', password='%s')>" % (self.name, self.fullname, self.password)
Base.metadata.create_all(engine)
u1= User(name='hawk', fullname='hawksnowlog', password='xxxxxxx')
Session = sessionmaker(bind=engine)
session = Session()
for i in session.query(User).order_by(User.id):
print(i.name)登録する部分を削って取得する部分を足します
いわゆる SELECT 文は session.query() を使います
とりあえず order_by を使っていますがデータは少ないので何でも OK です
SELECT した結果 User クラスの配列を返したい場合は session.query(User).order_by(User.id).all() を呼び出せば OK です
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
url = 'mysql+mysqldb://root@localhost/test?charset=utf8'
engine = create_engine(url, echo=True)
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String(50))
fullname = Column(String(100))
password = Column(String(100))
def __repr__(self):
return "<User(name='%s', fullname='%s', password='%s')>" % (self.name, self.fullname, self.password)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
u1 = session.query(User).get(1)
session.delete(u1)
session.commit()一旦 SELECT してからそのオブジェクトを削除します
あとは commit すれば OK です
削除とほぼ同じです
SELECT してそのオブジェクトの attribute に対してアクセスするだけです
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
url = 'mysql+mysqldb://root@localhost/test?charset=utf8'
engine = create_engine(url, echo=True)
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String(50))
fullname = Column(String(100))
password = Column(String(100))
def __repr__(self):
return "<User(name='%s', fullname='%s', password='%s')>" % (self.name, self.fullname, self.password)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
u1 = session.query(User).get(2)
u1.password = 'XXXXXXXX'
session.commit()最後に commit するのを忘れずに
Python3 で SQLAlchemy に入門してみました
慣れれば簡単に使えると思います
今回紹介した内容はかなり基本的なことだけです
もっと複雑なクエリの発行や外部キーの制約なども行えます
いろいろなサイトに書いてありましたが一番役に立つのは参考サイトにある公式サイトのチュートリアルなので、これをベースに勉強すると良いと思います
この記事もそのチュートリアルを元に作成しています
dockerhub で配布されている公式の swagger-ui イメージを使います
docker run -p 8080:8080 -v $(pwd)/swagger.yml:/usr/share/nginx/html/swagger.yml -e "API_URL=swagger.yml" swaggerapi/swagger-uiこれで localhost:8080 にアクセスすると確認できます
flasgger の Marshmallow Schemas のテクニックを紹介します
hoge = fields.Str(missing='TCP')hoge = fields.Str(enum=['A', 'B', 'C'])hoge = fields.Int(min=1)hoge = fields.Bool()hoge = fields.Str(dump_only=True)タイトルなどのトップレベルでのメタ情報を定義する方法です
app = Flask(__name__)
meta_info = {
'info': {
'description': 'description',
'title': 'title',
'contact': {
'name': 'https://hawksnowlog.blogspot.com'
},
'license': {
'name': 'hawksnowlog'
},
'version': '1.0.0',
'uiversion': 3,
'termsOfService': '/there_is_no_tos'
},
'tags': [
{
'name': 'tag1',
'description': 'tag1 description'
},
],
'schemes': ['https'],
'basePath': '/v1/api',
'host': 'localhost'
}
Swagger(app, template=meta_info)P.S 20180730 以下のように SwaggerView 側で array を定義することで可能だということがわかりました
class MyView(SwaggerView):
parameters = [
{
'in': 'body',
'name': 'HogeSpec parameter',
'description': 'Test array parameter',
'required': True,
'schema': {
'type': 'array',
'items': {
'$ref': '#/definitions/HogeSchema'
}
}
}
]調査中、以下のようなことがやりたいができない
class TestSchema(Schema):
fields.List(fields.Nested(HogeSchema))で
[
{
'key': 'value'
}
]という構造を作成したいが以下のようにしないと動作しない
class TestSchema(Schema):
fuga = fields.List(fields.Nested(HogeSchema)){
'fuga': [
{
'key': 'value'
}
]
}class MyView(SwaggerView):
tags = ['tag1']
summary = 'summary'
description = 'description'
operationId = 'operationId'
consumes = ['application/json']
produces = ['application/json']
parameters = [
{
'in': 'path',
'name': 'id',
'required': True,
'type': 'string',
}
]
responses = {
200: {
'description': 'OK',
'schema': TestSchema
}
}定義した View は add_url_rule で追加することで API として動作させます
app = Flask(__name__)
app.add_url_rule(
'/my',
view_func=MyView.as_view('my'),
methods=['GET']
)from flask import request
request.json.get('Hoge')どうやら SwaggerView 側の schema で参照しないと JSON には登場しません
なので
'schema': {
'type': 'array',
'items': {
'$ref': '#/definitions/HogeSchema'
}
}こんな感じでスキーマ参照しているとスキーマがないと言われてエラーとなります
そんな場合には definitions というパラメータで明示的に指定してあげることで JSON に登場させることができます
definitions = {
'HogeSchema': HogeSchema
}flasgger は内部的に apispec を使っています
apispec は OpenAPI-Specification ベースの定義ファイルをコードから生成することができるモジュールです
今回は flasgger 上のコードから apispec の機能を使って swagger の定義ファイルを生成してみました
pipenv install flasgger marshmallow apispec=="0.38.0"vim test_apispec.py# coding: utf-8
from flask import Flask, jsonify
from flasgger import APISpec, Schema, Swagger, fields
# Create an APISpec
spec = APISpec(
title='Flasger Petstore',
version='1.0.10',
plugins=[
'apispec.ext.flask',
'apispec.ext.marshmallow',
],
)
app = Flask(__name__)
class CategorySchema(Schema):
id = fields.Int()
name = fields.Str(required=True)
class PetSchema(Schema):
category = fields.Nested(CategorySchema, many=True)
name = fields.Str()
@app.route('/random')
def random_pet():
"""A cute furry animal endpoint.
---
get:
description: Get a random pet
responses:
200:
description: A pet to be returned
schema:
$ref: '#/definitions/Pet'
"""
pet = {'category': [{'id': 1, 'name': 'rodent'}], 'name': 'Mickey'}
return jsonify(PetSchema().dump(pet).data)
template = spec.to_flasgger(
app,
definitions=[CategorySchema, PetSchema],
paths=[random_pet]
)
swag = Swagger(app, template=template)
print(spec.to_dict())
print(spec.to_yaml())
if __name__ == '__main__':
app.run(debug=True)最後の print 文 2 つで dictionary 形式と YAML 形式の定義ファイルを出力しています
最初に APISpec を使ってメタデータなどを定義します
plugings は定義ファイルには表示されませんが必須です
spec = APISpec(
title='Flasger Petstore',
version='1.0.10',
plugins=[
'apispec.ext.flask',
'apispec.ext.marshmallow',
],
)その後に definitions 用のクラスの定義と paths に応じた処理が書いてあります
そしてその次に spec に対して定義した definitions と paths を登録します
template = spec.to_flasgger(
app,
definitions=[CategorySchema, PetSchema],
paths=[random_pet]
)更にその後で swag = Swagger(app, template=template) という設定がありますがこれは /apidocs を表示するための設定なのでなくても問題ないです
pipenv run python3 test_apispec.pyで
curl http://localhost:5000/randomという感じでリクエストすると動作します
また、サーバを動作させたターミナル上で dictionary と YAML の定義情報が表示されているのがわかると思います
definitions:
Category:
properties:
name: {type: string}
id: {format: int32, type: integer}
required: [name]
type: object
Pet:
properties:
name: {type: string}
category:
items: {$ref: '#/definitions/Category'}
type: array
type: object
info: {title: Flasger Petstore, version: 1.0.10}
parameters: {}
paths:
/random:
get:
description: Get a random pet
responses:
200:
description: A pet to be returned
schema: {$ref: '#/definitions/Pet'}
swagger: '2.0'
tags: []今回の定義だと上記のような感じで表示されると思います
flasgger の APISpec の機能を使ってコードから swagger ファイルを出力してみました
おそらくこれがあるのは swagger ファイルからモデルを生成するのではなくコードから swagger ファイルを生成したいという需要があるからだと思います
普通に考えれば swagger-codegen などを使ってコードを生成しますが、それだと内部がブラックボックスなのと結局コードと swagger ファイルをメンテンスしなければならないので、それであればコードだけを書き続けて swagger ファイルを自動生成するほうが良いということなんだと思います
flasgger には swagger ファイルをいろいろな形式で定義することができます
その中に Marshmallow Schemas というものがあります
Pythonのライブラリに marshmallow というものがありこれはオブジェクトのシリアライズを簡単に行うことができます
flasgger で marshmallow を使うでき、これを使うと Python でクラスを定義するように swagger を定義することができます
今回はサンプルを動かしてみました
pipenv install flasgger marshmallow apispec=="0.38.0"apispec というモジュールに依存しているのですが、これの最新版が 0.39.0 になっています
まだ flasgger 側が 0.39.0 に対応していないため一つ前のバージョンを明示的に指定します
Pull request はすでに出ているようでこれがマージされれば apispec の最新版が使えます
Marshmallow Schemas を使ったサンプルアプリを作成します
swagger 定義をコードに記載するので既存の定義ファイルは使いません
from flask import Flask, jsonify
from flasgger import Swagger, SwaggerView, Schema, fields
class Color(Schema):
name = fields.Str()
class Palette(Schema):
pallete_name = fields.Str()
colors = fields.Nested(Color, many=True)
class PaletteView(SwaggerView):
parameters = [
{
"name": "palette",
"in": "path",
"type": "string",
"enum": ["all", "rgb", "cmyk"],
"required": True,
"default": "all"
}
]
responses = {
200: {
"description": "A list of colors (may be filtered by palette)",
"schema": Palette
}
}
def get(self, palette):
"""
Colors API using schema
This example is using marshmallow schemas
"""
all_colors = {
'cmyk': ['cian', 'magenta', 'yellow', 'black'],
'rgb': ['red', 'green', 'blue']
}
if palette == 'all':
result = all_colors
else:
result = {palette: all_colors.get(palette)}
return jsonify(result)
app = Flask(__name__)
swagger = Swagger(app)
app.add_url_rule(
'/colors/<palette>',
view_func=PaletteView.as_view('colors'),
methods=['GET']
)
app.run(debug=True)definitions の役割は Schema クラスを継承した子クラスとして定義します
paramters と responses の部分は SwaggerView クラスを継承した子クラスとして定義します
ルーティングは app.add_url_rule で定義します
localhost:5000 で起動します
curl http://localhost:5000/colors/all/で動作します
/apidocs/ で swagger UI が確認できます
直接 swagger.yml の情報を Python のコードにも書けるのですが今回は YAML ファイルは分離して動かしてみたいと思います
parameters:
- name: palette
in: path
type: string
enum: ['all', 'rgb', 'cmyk']
required: true
default: all
definitions:
Palette:
type: object
properties:
palette_name:
type: array
items:
$ref: '#/definitions/Color'
Color:
type: string
responses:
200:
description: A list of colors (may be filtered by palette)
schema:
$ref: '#/definitions/Palette'
examples:
rgb: ['red', 'green', 'blue']from flask import Flask, jsonify
from flasgger import Swagger
from flasgger.utils import swag_from
app = Flask(__name__)
Swagger(app)
@app.route('/colors/<palette>/')
@swag_from('test.yml')
def index(palette):
all_colors = {
'cmyk': ['cian', 'magenta', 'yellow', 'black'],
'rgb': ['red', 'green', 'blue']
}
if palette == 'all':
result = all_colors
else:
result = {palette: all_colors.get(palette)}
return jsonify(result)
app.run(debug=True)単純な JSON を返却するだけのアプリです
定義した test.yml はデコレーションを使って参照します
@swag_from('test.yml')Swagger(app) することで swagger UI と JSON を参照できるようにしています
あとは @app.route('/colors/<palette>/') デコレーションでルーティングを定義します
<palette> は関数内で変数として参照することができます
localhost:5000 でアプリ起動します
curl http://localhost:5000/colors/all/で以下のレスポンスが返ってきます
{
"cmyk": [
"cian",
"magenta",
"yellow",
"black"
],
"rgb": [
"red",
"green",
"blue"
]
}http://localhost:5000/apidocs/ にアクセスすると swagger UI が表示されます
Python3 + flasgger を使って swagger.yml ファイルからアプリを作成してみました
これが基本的な使い方になると思います
少し気になったのはコントローラの分離方法です
やり方がわかったらこの辺りも紹介したいと思います
Fabric2 はデプロイツールで対象のサーバに SSH してコマンドを実行することができます
fabric を含むコードをテストする場合対象のサーバにアクセスできない状況などあると思います
その場合に unittest.Mock を使うことで擬似的に SSH してテストすることができます
今回はその方法を紹介します
vim my_fab.pyfrom fabric import Connection
class MyFab(object):
def __init__(self):
self.conn = Connection("root@172.28.128.3", connect_kwargs = { "password": "xxxxxxx" })
def call(self):
return self.conn.run('uname -s')
if __name__ == "__main__":
f = MyFab()
res = f.call()
print(res)これで実行すると当然ですが指定の IP アドレスに SSH 接続しにいきます
これを Mock 化して実際には SSH しないでテストしたいと思います
vim test_my_fab.pyimport unittest
from my_fab import MyFab
from unittest.mock import patch, MagicMock
class TestMyFabClass(unittest.TestCase):
@patch('my_fab.Connection')
def test_call(self, mock):
mock_res = MagicMock(return_value='Linux')
mock.return_value.run = mock_res
f = MyFab()
res = f.call()
print(res)
self.assertEqual(res, 'Linux')
if __name__ == '__main__':
unittest.main()ポイントは @patch('my_fab.Connection') です
fabric.Connection ではなく自分で作成したモジュールを指定します
あとは Connection はメソッドなので mock.return_value.run で返り値を設定します
fabric2 で Mock を使ったテスト方法を紹介しました
run と同じように put なども Mock 化すればテストできると思います
例えばライブリを使っている場合にどのモジュールから作成されたオブジェクトなのかわからなくなることがあると思います
そんな場合に便利かなと思います
class Car(object):
def __init__(self, name , color):
self.name = name
self.color = color
self.dist = 0
def run(self):
self.dist += 10
if __name__ == "__main__":
c = Car('fit', 'color')
print(c.__module__ + "." + c.__class__.__name__)最後の print(c.__module__ + "." + c.__class__.__name__) で表示しています
上記の場合は __main__.Car と表示されます
from car import Car
c = Car('fit', 'black')
print(c.__module__ + "." + c.__class__.__name__)これだと car.Car と表示されます
unittest は Python3 に標準で搭載されているテストモジュールです
簡単なクラスを作成してテストしてみました
import requests
import json
class TestClient(object):
def req_get(self):
return json.loads(requests.get('https://kaka-request-dumper.herokuapp.com/').text)
def req_post(self):
return json.loads(requests.post('https://kaka-request-dumper.herokuapp.com/').text)import unittest
from mock import TestClient
class TestMockClass(unittest.TestCase):
def test_get(self, mock):
cli = TestClient()
res = cli.req_get()
self.assertEqual(res['method'], 'GET')
def test_post(self):
cli = TestClient()
res = cli.req_post()
self.assertEqual(res['method'], 'POST')
if __name__ == '__main__':
unittest.main()python3 -m unittest test_mock.pyこれでテストすると実際にサーバにアクセスしてテストします
これを Mock 化してサーバにアクセスしないで各メソッドをテストします
import unittest
import json
from mock import TestClient
from unittest.mock import patch, MagicMock
class TestMockClass(unittest.TestCase):
@patch('mock.requests')
def test_get(self, mock):
mock_res = MagicMock(text=json.dumps({'method': 'GET'}))
mock.get.return_value = mock_res
cli = TestClient()
res = cli.req_get()
self.assertEqual(res['method'], 'GET')
@patch('mock.requests')
def test_post(self, mock):
mock_res = MagicMock(text=json.dumps({'method': 'POST'}))
mock.post.return_value = mock_res
cli = TestClient()
res = cli.req_post()
self.assertEqual(res['method'], 'POST')
if __name__ == '__main__':
unittest.main()MagicMock クラスを使ってモックを作成します
ポイントは以下の 2 行です
mock_res = MagicMock(text=json.dumps({'method': 'GET'}))
mock.get.return_value = mock_resMagicMock でモックが返却するレスポンスを作成します
今回は requests.get が JSON を返却して、その JSON 内の method というキーをチェックします
request.get.text で JSON 文字列が受け取れなければいけません
なので MagicMock(text=json.dumps({'method': 'GET'})) という感じでレスポンスを作成します
text= は request.get.text の最後の .text になります
.text を参照したときに JSON 文字列が返ってきますよという宣言になります
そして作成したレスポンスを mock.get.return_value = mock_res に設定します
mock オブジェクトはテスト用の関数 (test_get) の引数として受け取っています
これは何かと言うとモック化した requests になります
@patch('mock.requests') のデコレータを付与することでそれを実現しています
mock オブジェクト (requests) の get の return_value を先ほど作成したモックレスポンスとして設定しています
こうすることで実際にサーバにアクセスすることなくサーバと同等のレスポンスを返却する Mock を定義することができます
python3 -m unittest test_mock.pyこれで実行すると実際にサーバにアクセスせずテストが成功することが確認できると思います
1 つのテスト中で複数のメソッドやクラスに対して @patch を当てることも可能です
例えば今回のサンプルであれば
@patch('mock.requests')
@patch('mock.json')
def test_get(self, jmock, reqmock):こんな感じで複数のモジュールに対してモックを当たられます
複数のモックを受け取るためにテストに対して複数の引数を指定しましょう
上に指定したモックが後ろの引数に入る点に注意しましょう
unittest の Mock 機能を使ってみました
基本的にはレスポンスを Mock でエミュレートして擬似的なレスポンス情報を返却させる感じだと思います
MagicMock には他にもたくさんの機能があるので詳細は参考サイトにある公式のページを参考にしてください
unittest は Python3 に標準で搭載されているテストモジュールです
簡単なクラスを作成してテストしてみました
class Car(object):
def __init__(self, name , color):
self.name = name
self.color = color
self.dist = 0
def run(self):
self.dist += 10import unittest
from car import Car
class TestCarClass(unittest.TestCase):
@classmethod
def setUpClass(cls):
print('setUpClass')
def setUp(self):
print('setUp')
@classmethod
def tearDownClass(cls):
print('tearDownClass')
def tearDown(self):
print('tearDown')
def test_run(self):
c1 = Car('note', 'black')
c1.run()
self.assertEqual(c1.dist, 10)
def test_name(self):
self.c = Car('fit', 'silver')
self.assertEqual(self.c.name, 'fit')
if __name__ == '__main__':
unittest.main()python3 -m unittest test.pyテストを実行するときは -m で unittest を指定します
テストのクラスは unittest.TestCase を継承する必要があります
setUp と tearDown は各テストが実行される前と後で毎回呼ばれます
setUpClass と tearDownClass はこのテストが実行される際に最初と最後で一度だけ呼ばれます
テストには命名規則があり test_xxx という感じで test が先頭に付与される必要があります
他に使える assert は公式のページに記載があります
他にも minitest や pytest というライブラリがあります
標準の unittest もかなり機能はそろっているので正直これで十分なケースがほとんどかなと思います