最近这些年，REST已经成为web services和APIs的标准架构，很多APP的架构基本上是使用RESTful的形式了。
REST的六个特性：
- 客户端-服务器（Client-Server）服务器和客户端之间有明确的界限。一方面，服务器端不再关注用户界面和用户状态。另一方面，客户端不再关注数据的存储问题。这样，服务器端跟客户端可以独立开发，只要它们共同遵守约定。
- 无状态（Stateless）来自客户端的每个请求必须包含服务器所需要的所有信息，也就是说，服务器端不存储来自客户端的某个请求的信息，这些信息应由客户端负责维护。
- 可缓存（Cachable）服务器的返回内容可以在通信链的某处被缓存，以减少交互次数，提高网络效率。
- 分层系统（Layered System）允许在服务器和客户端之间通过引入中间层（比如代理，网关等）代替服务器对客户端的请求进行回应，而且这些对客户端来说不需要特别支持。
- 统一接口（Uniform Interface）客户端和服务器之间通过统一的接口（比如 GET, POST, PUT, DELETE 等）相互通信。
- 支持按需代码（Code-On-Demand，可选）服务器可以提供一些代码（比如 Javascript）并在客户端中执行，以扩展客户端的某些功能。
RESTful web service的样子
REST架构就是为了HTTP协议设计的。RESTful web services的核心概念是管理资源。资源是由URIs来表示，客户端使用HTTP当中的'POST, OPTIONS, GET, PUT, DELETE'等方法发送请求到服务器，改变相应的资源状态。
HTTP请求方法通常也十分合适去描述操作资源的动作：REST请求并不需要特定的数据格式，通常使用JSON作为请求体，或者URL的查询参数的一部分
使用flask 提供REST web服务
创建api蓝本
api蓝本结构
api/api_1_0/__init__.py API蓝本的构造文件
#！/usr/bin/env python
# -*- coding:utf-8 -*-
from flask import Blueprint api = Blueprint('api',__name__) from . import authentication,posts,users,comments,errors
app/__init__.py 注册API蓝本
def create_app(config_name):
from .api_1_0 import api as api_1_0_blueprint
app.register_blueprint(api_1_0_blueprint,url_prefix='/api/v1.0')
错误处理
app/main/errors.py 使用HTTP内容协商处理错误
#！/usr/bin/env python
# -*- coding:utf-8 -*-
from flask import render_template,request,jsonify
from . import main #主程序的errorhandler
@main.app_errorhandler(404)
def page_not_find(e):
# 程序检查Accept请求首部request.accept_mimetypes，根据首部的值决定客户端期望接收的响应格式
if request.accept_mimetypes.accept_json and not request.accept_mimetypes.accept_html:
response = jsonify({'error':'not found'})
response.status_code =404
return response
return render_template('404.html'),404 @main.app_errorhandler(403)
def forbidden(e):
return render_template('403.html'),403 @main.app_errorhandler(500)
def internal_server_error(e):
if request.accept_mimetypes.accept_json and not request.accept_mimetypes.accept_html:
response = jsonify({'error': 'internal server error'})
response.status_code = 500
return response
return render_template('500.html'),500
app/api_1_0/errors.py API蓝本中错误处理程序
#！/usr/bin/env python
# -*- coding:utf-8 -*-
from flask import jsonify def forbidden(message):
response = jsonify({'error':'forbidden','message':message})
response.status_code=403
return response def bad_request(message):
response = jsonify({'error': 'bad request', 'message': message})
response.status_code = 400
return response def unauthorized(message):
response = jsonify({'error': 'unauthorized', 'message': message})
response.status_code = 401
return response
使用Flask-HTTPauth认证用户
REST架构基于HTTP协议，发送密令的最佳方式时HTTP认证，用户密令包含在请求的Authorization首部中
pip install flask-httpauth
app/api_1_0/authorization.py 初始化HTTPauth
#！/usr/bin/env python
# -*- coding:utf-8 -*-
from flask_httpauth import HTTPBasicAuth
from flask import g
from ..models import AnonymousUser,User auth = HTTPBasicAuth() @auth.verify_password
def vertify_password(email,password):
# 匿名访问，邮件字段为空
if email == '':
# g为flask的全局对象
g.current_user=AnonymousUser()
return True
user = User.query.filter_by(email=email).first()
if not user:
return False
g.current_user=user
return user.verify_password(password)
如果密令认证不正确，为了返回和其他API返回的错误一致，需要自定义错误响应
from .errors import unauthorized,forbidden @auth.error_handler
def auth_error():
return unauthorized('无效认证')
为保护路由可以使用修饰器 auth.login_required，可注释掉
@api.route('/posts/')
@auth.login_required
def get_posts():
pass
蓝本中所有的路由都需要使用相同的方式进行保护，所以在before_request 处理程序中使用一次login_required 修饰器，应用到整个蓝本
app/api_1_0/authorization.py before_request 处理程序中进行认证
from .errors import unauthorized,forbidden @api.before_request
@auth.login_required
def before_request():
if not g.current_user.is_anonymous and not g.current_user.confirmed:
return forbidden('账户未确认')
基于令牌的认证
app/models.py 支持基于令牌的认证
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
class User(UserMixin,db.Model):
def generate_auth_token(self,expiration):
# 生成验证Token
s= Serializer(current_app.config['SECRET_KEY'],expires_in=expiration)
return s.dumps({'id':self.id}).decode('ascii') # 需要编码，否则报错
@staticmethod # 因为只有解码后才知道用户是谁，所以用静态方法 def verify_auth_token(token): # 验证token s = Serializer(current_app.config['SECRET_KEY']) try: data = s.loads(token) except: return None return User.query.get(data['id'])
app/api_1_0/authentication.py 支持令牌的改进验证回调
@auth.verify_password
def vertify_password(email_or_token,password):
if email_or_token == '':
g.current_user=AnonymousUser()
return True
if password =='':
# 如果密码为空假定参数时令牌，按照令牌去认证
g.current_user = User.verify_auth_token(email_or_token)
# 为避免客户端用旧令牌申请新令牌，如果使用令牌认证就拒绝请求
g.token_used = True
return g.current_user is not None
user = User.query.filter_by(email=email_or_token).first()
if not user:
return False
g.current_user=user
# 为了让视图函数区分两种认证方法 添加了token_used变量
g.token_used = False
return user.verify_password(password)
app/api_1_0/authentication.py 生成认证令牌
# 生成认证令牌
@api.route('/token')
def get_token():
# 为避免客户端用旧令牌申请新令牌，如果使用令牌认证就拒绝请求
if g.current_user.is_anonymous() or g.token_used:
return unauthorized('无效认证')
return jsonify({'token':g.current_user.generate_auth_token(expiration=3600),'expiration':3600})
资源和JSON的序列化转换
app/models.py 把文章转换成JSON格式化序列化的字典
class Post(db.Model):
def to_json(self):
json_post = {
'url':url_for('api.get_post',id = self.id,_external=True),
'body':self.body,
'body_html':self.body_html,
'timestamp':self.timestamp,
'author':url_for('api.get_user',id=self.author_id,_external=True),
'comments':url_for('api.get_post_comments',id = self.id,_external=True),
'comments_count':self.comments.count()
}
return json_post
app/models.py 把用户转换成JSON格式化字典
class User(UserMixin,db.Model):
def to_json(self):
json_user = {
'url':url_for('api.get_post',id = self.id,_external=True),
'body':self.username,
'member_since':self.member_since,
'last_seen':self.last_seen,
'posts':url_for('api.get_user_posts',id=self.id,_external=True),
'followed_posts':url_for('api.get_user_followed_posts',id = self.id,_external=True),
'posts_count':self.posts.count()
}
return json_user
app/models.py 从JSON格式数据创建一篇博客文章
from app.exceptions import ValidationError
class Post(db.Model):
@staticmethod
def form_json(json_post):
body = json_post.get('body')
if body is None or body=='':
raise ValidationError('文章没有body字段')
return Post(body=body)
app.exceptions.py
class ValidationError(ValueError):
pass
app/api_1_0/errors.py API中ValidationError 异常的处理程序
# 全局异常处理程序，只有从蓝本中的路由抛出异常才会调用处理这个程序
@api.errorhandler(ValidationError)
def validation_error(e):
return bad_request(e.args[0])
app/api_1_0/posts.py 文章资源的GET请求处理程序
@api.route('/posts/')
def get_posts():
page = request.args.get('page', 1, type=int)
pagination = Post.query.paginate(
page, per_page=current_app.config['FLASKY_POSTS_PER_PAGE'],
error_out=False)
posts = pagination.items
prev = None
if pagination.has_prev:
prev = url_for('api.get_posts', page=page-1, _external=True)
next = None
if pagination.has_next:
next = url_for('api.get_posts', page=page+1, _external=True)
return jsonify({
'posts': [post.to_json() for post in posts],
'prev': prev,
'next': next,
'count': pagination.total
})
@api.route('/posts/<int:id>')
def get_post(id):
post = Post.query.get_or_404(id)
return jsonify(post.to_json())
app/api_1_0/posts.py 文章资源的POST请求处理程序
@api.route('/posts/', methods=['POST'])
@permission_required(Permission.WRITE_ARTICLES)
def new_post():
post = Post.from_json(request.json)
post.author = g.current_user
db.session.add(post)
db.session.commit()
return jsonify(post.to_json()), 201, \
{'Location': url_for('api.get_post', id=post.id, _external=True)}
app/api_1_0/decorators.py permisson_required 修饰器
#！/usr/bin/env python
# -*- coding:utf-8 -*-
from functools import wraps
from flask import g
from .errors import forbidden def permission_required(permission):
def decorator(f):
@wraps(f)
def decorated_function(*args,**kwargs):
if not g.current_user.can(permission):
return forbidden('无权限')
return f(*args,**kwargs)
return decorated_function
return decorator
app/api_1_0/posts.py 文章资源PUT请求处理程序
# 更新现有资源
@api.route('/posts/<int:id>', methods=['PUT'])
@permission_required(Permission.WRITE_ARTICLES)
def edit_post(id):
post = Post.query.get_or_404(id)
if g.current_user != post.author and \
not g.current_user.can(Permission.ADMINISTER):
return forbidden('无权限')
# 更新body
post.body = request.json.get('body', post.body)
db.session.add(post)
return jsonify(post.to_json())
app/api_1_0/users.py
#！/usr/bin/env python
# -*- coding:utf-8 -*-
from flask import jsonify, request, current_app, url_for
from . import api
from ..models import User, Post @api.route('/users/<int:id>')
def get_user(id):
user = User.query.get_or_404(id)
return jsonify(user.to_json()) @api.route('/users/<int:id>/posts/')
def get_user_posts(id):
user = User.query.get_or_404(id)
page = request.args.get('page', 1, type=int)
pagination = user.posts.order_by(Post.timestamp.desc()).paginate(
page, per_page=current_app.config['FLASKY_POSTS_PER_PAGE'],
error_out=False)
posts = pagination.items
prev = None
if pagination.has_prev:
prev = url_for('api.get_user_posts', page=page-1, _external=True)
next = None
if pagination.has_next:
next = url_for('api.get_user_posts', page=page+1, _external=True)
return jsonify({
'posts': [post.to_json() for post in posts],
'prev': prev,
'next': next,
'count': pagination.total
}) @api.route('/users/<int:id>/timeline/')
def get_user_followed_posts(id):
user = User.query.get_or_404(id)
page = request.args.get('page', 1, type=int)
pagination = user.followed_posts.order_by(Post.timestamp.desc()).paginate(
page, per_page=current_app.config['FLASKY_POSTS_PER_PAGE'],
error_out=False)
posts = pagination.items
prev = None
if pagination.has_prev:
prev = url_for('api.get_user_followed_posts', page=page-1,
_external=True)
next = None
if pagination.has_next:
next = url_for('api.get_user_followed_posts', page=page+1,
_external=True)
return jsonify({
'posts': [post.to_json() for post in posts],
'prev': prev,
'next': next,
'count': pagination.total
})
app/api_1_0/comments.py
#！/usr/bin/env python
# -*- coding:utf-8 -*-
from flask import jsonify, request, g, url_for, current_app
from .. import db
from ..models import Post, Permission, Comment
from . import api
from .decorators import permission_required @api.route('/comments/')
def get_comments():
page = request.args.get('page', 1, type=int)
pagination = Comment.query.order_by(Comment.timestamp.desc()).paginate(
page, per_page=current_app.config['FLASKY_COMMENTS_PER_PAGE'],
error_out=False)
comments = pagination.items
prev = None
if pagination.has_prev:
prev = url_for('api.get_comments', page=page-1, _external=True)
next = None
if pagination.has_next:
next = url_for('api.get_comments', page=page+1, _external=True)
return jsonify({
'comments': [comment.to_json() for comment in comments],
'prev': prev,
'next': next,
'count': pagination.total
}) @api.route('/comments/<int:id>')
def get_comment(id):
comment = Comment.query.get_or_404(id)
return jsonify(comment.to_json()) @api.route('/posts/<int:id>/comments/')
def get_post_comments(id):
post = Post.query.get_or_404(id)
page = request.args.get('page', 1, type=int)
pagination = post.comments.order_by(Comment.timestamp.asc()).paginate(
page, per_page=current_app.config['FLASKY_COMMENTS_PER_PAGE'],
error_out=False)
comments = pagination.items
prev = None
if pagination.has_prev:
prev = url_for('api.get_post_comments', id=id, page=page-1,
_external=True)
next = None
if pagination.has_next:
next = url_for('api.get_post_comments', id=id, page=page+1,
_external=True)
return jsonify({
'comments': [comment.to_json() for comment in comments],
'prev': prev,
'next': next,
'count': pagination.total
}) @api.route('/posts/<int:id>/comments/', methods=['POST'])
@permission_required(Permission.COMMENT)
def new_post_comment(id):
post = Post.query.get_or_404(id)
comment = Comment.from_json(request.json)
comment.author = g.current_user
comment.post = post
db.session.add(comment)
db.session.commit()
return jsonify(comment.to_json()), 201, \
{'Location': url_for('api.get_comment', id=comment.id,
_external=True)}
使用HTTPie测试web服务
pip install httpie
http --json --auth 834424581@qq.com:abc GET http://127.0.0.1:5000/api/v1.0/posts
匿名用户，空邮箱，空密码
http --json --auth : GET http://127.0.0.1:5000/api/v1.0/posts
POST 添加文章
http --auth 834424581@qq.com:abc --json POST http://127.0.0.1:5000/api/v1.0/posts/ “body=xxxxxxxxxxxxxxxx”
使用认证令牌，可以向api/v1.0/token发送请求
http --auth 834424581@qq.com:abc --json GET http://127.0.0.1:5000/api/v1.0/token
接下来的1小时，可以用令牌空密码访问
http --auth eyJpYXQ......: --json GET http://127.0.0.1:5000/api/v1.0/posts
令牌过期，请求会返回401错误，需要重新获取令牌