通过 Flask Restful Api 对model进行CRUD （create,read,update,delete）

开始

源代码 here

$ git clone https://github.com/rahmanfadhil/flask-rest-api.git

创建venv，安装依赖

$ python -m venv env
$ source env/bin/activate
(env) $ pip install -r requirements.txt

依赖

• Flask SQLAlchemy: A Flask extension that adds support for SQLAlchemy, 关联数据库
• Flask RESTful: 创建restful接口
• Flask Marshmallow: 序列化对象方法Marshmallow

step1

$ python3 -m venv env
$ source env/bin/activate

$ pip install Flask \
    Flask-SQLAlchemy \
    Flask-RESTful \
    flask-marshmallow

创建主函数 main.py

from flask import Flask

app = Flask(__name__)

if __name__ == '__main__':
    app.run(debug=True)

测试一下

(env) $ python main.py

Open http://localhost:5000, and you will see a 404 page.

数据库

这里选用sqlite，记得生产用postsql或mysql

from flask import Flask
from flask_sqlalchemy import SQLAlchemy # new

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db' # new
db = SQLAlchemy(app) # new

if __name__ == '__main__':
    app.run(debug=True)

Database model

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(50))
    content = db.Column(db.String(255))

    def __repr__(self): # str(post) 时会调用，即print时显示的string
        return '<Post %s>' % self.title

在restful api中返回的数据是json格式，所以需要把model转换为json，这就需要
flask_marshmallow ，先定义schema，再通过schema转化为json

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow # new

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db'
db = SQLAlchemy(app)
ma = Marshmallow(app) # new

# ...

if __name__ == '__main__':
    app.run(debug=True)

定义了两个schema实例

class PostSchema(ma.Schema):
    class Meta:
        fields = ("id", "title", "content")

post_schema = PostSchema()
posts_schema = PostSchema(many=True)

初始化数据库，当然也可以使用flask migrate，这里使用最简单的方式。

(env) $ python
>>> from main import db
>>> db.create_all()
>>> exit()

RESTful 路由

Get Posts api

class PostListResource(Resource):
    def get(self):
        posts = Post.query.all()
        return posts_schema.dump(posts) # schema 作用就是将posts转为dict

api.add_resource(PostListResource, '/posts') # 绑定了路由

测试

$ curl http://localhost:5000/posts
[]

Create Api

# ...
from flask import Flask, request # change

# ...

class PostListResource(Resource):
    def get(self):
        posts = Post.query.all()
        return posts_schema.dump(posts)

    # create api
    def post(self):
        new_post = Post(
            title=request.json['title'],
            content=request.json['content']
        )
        db.session.add(new_post)
        db.session.commit()
        return post_schema.dump(new_post)

# ...

测试

$ curl http://localhost:5000/posts \
    -X POST \
    -H "Content-Type: application/json" \
    -d '{"title":"Post 1", "content":"Lorem ipsum"}'
{
    "content": "Lorem ipsum",
    "id": 1,
    "title": "Post 1"
}

根据id获取post，get_or_404: 在获取不到时，返回404

class PostResource(Resource):
    def get(self, post_id):
        post = Post.query.get_or_404(post_id)
        return post_schema.dump(post)

api.add_resource(PostResource, '/posts/<int:post_id>')

测试1

$ curl http://localhost:5000/posts/1
{
    "title": "Post 1",
    "content": "Lorem ipsum",
    "id": 1
}

测试2

$ curl http://localhost:5000/posts/12 -I
HTTP/1.0 404 NOT FOUND
...

新增 update和delete

class PostResource(Resource):
    # ...

    def patch(self, post_id):
        post = Post.query.get_or_404(post_id)

        if 'title' in request.json:
            post.title = request.json['title']
        if 'content' in request.json:
            post.content = request.json['content']

        db.session.commit()
        return post_schema.dump(post)

    def delete(self, post_id):
        post = Post.query.get_or_404(post_id)
        post.delete()
        db.session.commit()
        return '', 204

测试 Update :

$ curl http://localhost:5000/posts/1 \
    -X PATCH \
    -H "Content-Type: application/json" \
    -d '{"title":"Updated Post", "content":"Updated post content"}'
{
    "content": "Updated post content",
    "id": 1,
    "title": "Updated Post"
}

** 测试 Delete :**

$ curl http://localhost:5000/posts/1 -X DELETE -I
HTTP/1.0 204 NO CONTENT
...