Flask 学习十三应用编程接口

时间 2019-11-11

标签 flask 学习十三应用编程接口栏目 Python 繁體版

原文原文链接

最近这些年，REST已经成为web services和APIs的标准架构，不少APP的架构基本上是使用RESTful的形式了。html

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'等方法发送请求到服务器，改变相应的资源状态。python

HTTP请求方法一般也十分合适去描述操做资源的动做：REST请求并不须要特定的数据格式，一般使用JSON做为请求体，或者URL的查询参数的一部分web

使用flask 提供REST web服务json

建立api蓝本flask

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内容协商处理错误session

#！/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错误，须要从新获取令牌

Flask 学习 十三 应用编程接口

REST的六个特性：

RESTful web service的样子

Flask 学习十三应用编程接口