Flask SQLAlchemy RESTful

超簡単Flask + SQLAlchemyでRESTful APIサーバー

September 17, 2019

#SQLAlchemy #Flask #APIサーバー #RESTful #python #Marshmallow #SQLite

Flask + SQLAlchemyで超簡単RESTful APIサーバー

1. 環境作成

 $ sudo apt install python3-pip python3-venv python3-dev sqlite3
 $ mkdir ~/myapp; cd ~/myapp
 $ python3 -m venv env
 $ source env/bin/activate
 (env) $ pip install flask_sqlalchemy flask_marshmallow

2. ファイル作成

~/myapp/server.py

 from flask import Flask, request
 from flask_sqlalchemy import SQLAlchemy
 from flask_marshmallow import Marshmallow
 
 
 app = Flask(__name__)
 app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///myapp.sqlite'
 db = SQLAlchemy(app)
 ma = Marshmallow(app)
 
 
 class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True)
    email = db.Column(db.String(120), unique=True)
 
    def __init__(self, username, email):
        self.username = username
        self.email = email
 
 
 class UserSchema(ma.Schema):
    class Meta:
        fields = ('username', 'email')
 
 
 user_schema = UserSchema()
 users_schema = UserSchema(many=True)
 
 
 @app.route("/user", methods=["POST"])
 def add_user():
    username = request.json['username']
    email = request.json['email']
    
    new_user = User(username, email)
 
    db.session.add(new_user)
    db.session.commit()
 
    return user_schema.jsonify(new_user) 
 
 
 @app.route("/user", methods=["GET"])
 def get_user():
    all_users = User.query.all()
 
    return users_schema.jsonify(all_users)
 
 
 @app.route("/user/<id>", methods=["GET"])
 def user_detail(id):
    user = User.query.get(id)
 
    return user_schema.jsonify(user)
 
 
 @app.route("/user/<id>", methods=["PUT"])
 def user_update(id):
    user = User.query.get(id)
    username = request.json['username']
    email = request.json['email']
 
    user.email = email
    user.username = username
 
    db.session.commit()
    return user_schema.jsonify(user)
 
 
 @app.route("/user/<id>", methods=["DELETE"])
 def user_delete(id):
    user = User.query.get(id)
    db.session.delete(user)
    db.session.commit()
 
    return user_schema.jsonify(user)
 
 
 if __name__ == '__main__':
    app.run(debug=True)

3. データベース作成&実行

 (env) $ python
 >>> from server import db
 >>> db.create_all()
 
 (env) $ python server.py

4. 動作確認

 # 追加
 $ curl -X POST -H 'Content-Type:application/json' -d '{"username":"name1","email":"email1"}' http://127.0.0.1:5000/user
 # 更新
 $ curl -X PUT -H 'Content-Type:application/json' -d '{"username":"name2","email":"email2"}' http://127.0.0.1:5000/user/1
 # 取得(1件)
 $ curl -X GET -H 'Content-Type:application/json' http://127.0.0.1:5000/user/1
 # 取得(全件)
 $ curl -X GET -H 'Content-Type:application/json' http://127.0.0.1:5000/user
 # 削除
 $ curl -X DELETE -H 'Content-Type:application/json' http://127.0.0.1:5000/user/1

Flask-Restlessを使うともっと簡単にいけそうでしたが、開発が止まってるようなのでSQLAlchemyMarshmallowだけで。