この記事はニフティグループ Advent Calendar 2018の10日目の穴を埋めるための記事です。
時空が捻じ曲がって昨日は@ma2さんのエンジニアへの俳句のすゝめだったようです。
初めに
業務でAPI作る際にFlask触ることがあったので投稿。
使ってみると軽量なフレームワークだったのでdjangoと比べて非常に取っ付きやすく構成も自由でいい感じでした。
ただFlask-Extentionsの設定やModelを作りこんでいくとバッチ処理でも共通で使いたくなってきます。manage.pyみたいな。。。
公式を漁っているとCommand Line Interface/Custom Commandsという機能があるようで、Clickというコマンドラインインタフェースベースでカスタムコマンド(=バッチ)が作れるみたいです。
http://flask.pocoo.org/docs/1.0/cli/#custom-commands
本投稿では実際に簡単なAPIを作った上で、Custom Commandを実装してみます。
Restful-APIを作ってみる
まずは簡単なユーザ登録・参照APIを作ってみます。ディレクトリ構成は以下の通り。
project_name/
├── apis/
| ├── __init__.py
│ └── user.py
├── models/
│ └── __init__.py
├── app.py
└── settings.py
Flask拡張としては有名なflask-restplusとflask-sqlalchemyを今回は使います。
環境
- Python 3.7
- flask 1.0.2
- MySQL 5.5
作成
まずは、database modelから実装してみます。
今回はひとつしかテーブルを作らないのでmodelはmodels/__init__.pyに全部書いてしまいます。
プロジェクトが大きくなったら公式を参考に分割する構造にしたほうがいいかと思います。
先々で必要になりそうな処理をmodelでは書いていきます。
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
class User(db.Model):
__tablename__ = 't_user'
id = db.Column(db.Integer, primary_key=True, autoincrement=False)
name = db.Column(db.String(length=32), nullable=False)
age = db.Column(db.Integer, nullable=False)
@classmethod
def create(cls, id, name, age):
"""
ユーザを作成する
:param id:
:param name:
:param age:
:return: User
"""
obj = cls(id=id, name=name, age=age)
db.session.add(obj)
return obj
@classmethod
def get_by_id(cls, id):
"""
IDを指定してユーザを取得
:param id:
:return: User
"""
return db.session.query(cls).filter(cls.id == id).one_or_none()
@classmethod
def get_all(cls):
"""
全ユーザを取得
:return: Userリスト
"""
return db.session.query(cls).all()
def increment_age(self, num=1):
"""
年齢を増やす
:return: None
"""
self.age = self.age+num
次に、APIエンドポイントを書いていきます。
今回はBlueprintを使わない方法でAPIエンドポイントごとにファイル分割できる構成にします。
flask-restplusを使っていますがメインの機能は殆ど使っていません。
flask-restplusの詳しい使い方は以下の記事をみて参考にしてください。
API仕様書を作らなくてもいい！？簡単にAPIを作成する方法
from models import db, User
from flask_restplus import Resource
from flask import request
class UserReq(Resource):
def get(self, id):
"""
指定されたIDのユーザ情報を返却する
:param id: 整数
:return: Response
"""
user = User.get_by_id(id)
if not user:
return {}, 404
return {
"id": user.id,
"name": user.name,
"age": user.age
}, 200
def put(self, id):
"""
指定されたIDでユーザを登録する
:param id: 整数
:return: Response
"""
payload = request.json
name = payload["name"]
age = payload["age"]
try:
User.create(id, name, age)
db.session.commit()
except:
db.session.rollback()
return {"result": False}, 500
return {"result": True}, 201
APIエンドポイントがひとつしかないのでさびしいですが、プロジェクトが大きくなってくると圧倒的にこっちの構成のほうがいいです。
from flask_restplus import Api
from .user import UserReq
api = Api()
# APIエンドポイントが増えると追加していく
api.add_resource(UserReq, '/users/<id>')
次にflask拡張関連のconfigを書きます。
データベースへの接続設定は各自の環境に合わせてください。
class BaseConfig:
SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://user:pass@host/database?charset=utf8'
最後にapp.pyを書きます。
import settings
from flask import Flask
from models import db
from apis import api
app = Flask(__name__)
app.config.from_object(settings.BaseConfig)
db.init_app(app)
api.init_app(app)
db.create_all(app=app)
if __name__ == '__main__':
app.run()
実行
環境にもよりますが、以下のコマンドで実行。
# flask run app.py
登録と参照のリクエストを投げてみる
# curl -H "Content-type: application/json" -X PUT -d '{"name":"Taro", "age":10}' http://localhost:5000/users/1
{"result": true}
# curl -H "Content-type: application/json" -X GET http://localhost:5000/users/1
{"id": 1, "name": "Taro", "age": 10}
動作には問題なさそうです。DBを直接見た結果は以下の通り。
mysql> select * from t_user;
+----+------+-----+
| id | name | age |
+----+------+-----+
| 1 | Taro | 10 |
+----+------+-----+
本題: Custom Command
ようやく本題です。
ここで先程のModelを使いまわすようなコマンド(=バッチ)を作りたくなると思います。
以下の3つのコマンドを作成してみようと思います。
1. "Hello World"コマンド
2. 全員の年齢を上げるコマンド
3. 大人か判断するコマンド
project_name/
├── apis/
| ├── __init__.py
│ └── user.py
├── jobs/
| ├── __init__.py ★
| ├── task1.py ★"Hello World"コマンド の本体
| ├── task2.py ★全員の年齢を上げるコマンド の本体
│ └── task3.py ★大人か判断するコマンド の本体
├── models/
│ └── __init__.py
├── app.py
├── cli.py ★
└── settings.py
★先程のプロジェクトに追加するファイル
作成1
まずは"Hello World"コマンドだけ作ってみます。
コマンドの本体から作成していきます。
import click
from flask.cli import with_appcontext
@click.command('task1', help="Hello World.")
@with_appcontext
def task1_run():
print("hello world!!!")
次に初期化をします。コマンド(=バッチ)が増える場合追記していくようにすると管理が楽です。
rom flask.cli import AppGroup
from jobs.task1 import task1_run
# グループを作成
job = AppGroup('job')
# task関連のコマンドを追加していく
job.add_command(task1_run)
最後に呼び出し元
import settings
from flask import Flask
from models import db
from jobs import job
from apis import api
app = Flask(__name__)
app.config.from_object(settings.BaseConfig)
db.init_app(app)
api.init_app(app)
app.cli.add_command(job)
実行1
早速実行してみます。
# export FLASK_APP=cli.py
# flask job task1
hello world!!!
問題なく動作しているようです。
作成2
次に全員の年齢を上げるコマンドを作ってみます。
こちらはAPIで使っていたModelを使いまわしてみます。
with_appcontextデコレータをつける事によってFlask拡張を使えるようになります。
import click
from flask.cli import with_appcontext
from models import db, User
# バッチ関連
@click.command('task2', help="All User Happy Birthday!!!")
@with_appcontext
def task2_run():
users = User.get_all()
if not users:
print("Not found user...")
else:
for user in users:
try:
user.increment_age()
db.session.commit()
print("Happy {}th Birthday!!! {}!!!".format(user.age, user.name))
except:
db.session.rollback()
print("Unhappy Birthday...")
初期化にも追加。
from flask.cli import AppGroup
from jobs.task1 import task1_run
from jobs.task2 import task2_run
# グループを作成
job = AppGroup('job')
# task関連のコマンドを追加
job.add_command(task1_run)
job.add_command(task2_run)
実行2
実行の前に、あらかじめ以下のユーザをAPIで登録しておきます。
+----+--------+-----+
| id | name | age |
+----+--------+-----+
| 1 | Taro | 10 |
| 2 | Hiro | 20 |
| 3 | Tomoko | 30 |
+----+--------+-----+
実行してみます。
# export FLASK_APP=cli.py
# flask job task2
Happy 11th Birthday!!! Taro!!!
Happy 21th Birthday!!! Hiro!!!
Happy 31th Birthday!!! Tomoko!!!
全員年齢が1歳上がりました！
+----+--------+-----+
| id | name | age |
+----+--------+-----+
| 1 | Taro | 11 |
| 2 | Hiro | 21 |
| 3 | Tomoko | 31 |
+----+--------+-----+
作成3
最後に大人か判断するコマンドを作ってみます。
Clickの特徴としてて様々なコマンドライン機能が簡単に実装できます。
下記のようにclick.optionデコレータで指定してあげればオプション引数が実装できます。
デフォルトも指定できます。
他にも機能がたくさんあるので、もっと知りたい人は公式を見てください。
https://click.palletsprojects.com/en/7.x/
import click
from flask.cli import with_appcontext
from models import db, User
# バッチ関連
@click.command('task3', help="")
@click.option('--age', type=click.INT, default=20)
@with_appcontext
def task3_run(age):
users = User.get_all()
for user in users:
if age <= user.age:
print("{}({}) is adult.".format(user.name, user.age))
else:
print("{}({}) is child.".format(user.name, user.age))
初期化にも追加。
from flask.cli import AppGroup
from jobs.task1 import task1_run
from jobs.task2 import task2_run
from jobs.task3 import task3_run
# グループを作成
job = AppGroup('job')
# task関連のコマンドを追加
job.add_command(task1_run)
job.add_command(task2_run)
job.add_command(task3_run)
実行3
そのまま実行
# export FLASK_APP=cli.py
# flask job task3
Taro(11) is child.
Hiro(21) is adult.
Tomoko(31) is adult.
オプション引数もばっちり受け取れます。
# export FLASK_APP=cli.py
# flask job task3 --age 25
Taro(11) is child.
Hiro(21) is child.
Tomoko(31) is adult.
最後に
今回はFlaskでバッチ処理ができるか試してみました。
Custom CommandはFlask拡張も簡単に使い回すことができる上に自由な構成で組めるのがいいですね。
ぜひ皆さんも使ってみてください。