Flask 學習 十三 應用編程接口

最近這些年，REST已經成爲web services和APIs的標準架構，不少APP的架構基本上是使用RESTful的形式了。

REST的六個特性：

1. 客戶端-服務器（Client-Server）服務器和客戶端之間有明確的界限。一方面，服務器端再也不關注用戶界面和用戶狀態。另外一方面，客戶端再也不關注數據的存儲問題。這樣，服務器端跟客戶端能夠獨立開發，只要它們共同遵照約定。
2. 無狀態（Stateless）來自客戶端的每一個請求必須包含服務器所須要的全部信息，也就是說，服務器端不存儲來自客戶端的某個請求的信息，這些信息應由客戶端負責維護。
3. 可緩存（Cachable）服務器的返回內容能夠在通訊鏈的某處被緩存，以減小交互次數，提升網絡效率。
4. 分層系統（Layered System）容許在服務器和客戶端之間經過引入中間層（好比代理，網關等）代替服務器對客戶端的請求進行迴應，並且這些對客戶端來講不須要特別支持。
5. 統一接口（Uniform Interface）客戶端和服務器之間經過統一的接口（好比 GET, POST, PUT, DELETE 等）相互通訊。
6. 支持按需代碼（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錯誤，須要從新獲取令牌