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

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":"name１","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
# 取得(１件)
$ 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/１

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