通過 Flask Restful Api 對model進行CRUD （create,read,update,delete）
開始
源代碼 here
$ git clone https://github.com/rahmanfadhil/flask-rest-api.git
創(chuàng)建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, 關聯(lián)數(shù)據(jù)庫
- Flask RESTful: 創(chuàng)建restful接口
- Flask Marshmallow: 序列化對象方法Marshmallow
step1
$ python3 -m venv env
$ source env/bin/activate
$ pip install Flask \
Flask-SQLAlchemy \
Flask-RESTful \
flask-marshmallow
創(chuàng)建主函數(shù) 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.
數(shù)據(jù)庫
這里選用sqlite，記得生產(chǎn)用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) 時會調(diào)用漓摩，即print時顯示的string
return '<Post %s>' % self.title
在restful api中返回的數(shù)據(jù)是json格式肢执，所以需要把model轉(zhuǎn)換為json扩灯，這就需要
flask_marshmallow 晕换，先定義schema学少，再通過schema轉(zhuǎn)化為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)
初始化數(shù)據(jù)庫，當然也可以使用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轉(zhuǎn)為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"
}
根據(jù)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
...
