Flask Rest接口

 Flask適用於簡單的接口請求

安裝 

pip install Flask

pip install Flask-RESTful

 僅簡單請求url，而後出發處理程序，返回處理結果

 

app.py代碼以下

from flask import Flask,jsonify

from flask_restful import Api,Resource,request,reqparse

from sqlalchemy import create_engine

from local_settings import SQLALCHEM

import logging,datetime

 

logger=logging.getLogger('my_app')

app=Flask(__name__)

api=Api(app)

class ProceListAPI(Resource):
    def __init__(self):
        self.engine=create_engine(alchemy_conf,pool_size=5,max_overflow=0,pool_recycle=7*60)     # 初始化全局的數據庫鏈接池

        self.reqparse=reqparse.RequestParser()

        self.reqparse.add_argument('subid',action='append',dest='cart_code',type=int,required=True,help='subid is needed!',location='form')

        self.reqparse.add_argument('description',type=str,default="",location='json')      # 配置 哪些入參是什麼格式，在什麼位置，dest表示別名

        supper(PriceListAPI,self).__init__()

    def get(self):     #flask的api類中的全部get請求通過 get方法 ，post請求 通過post方法

        logger.warning('request form:{},num:{}'.format(request.args,type(request.args.get('subid'))))

        cart_code = request.args.get('subid')

        if not cart_code:
            return {'error':'subid is Rquired!'}

        cart_code=cart_code.split(',')

        with self.engine.connect() as conn:

            result = conn.execute('select distinct cart_code,price from price_store where collect_time>=curdate();')

            datas=result.fetchall()

        dup_dict={}

        for d in datas:
            if d[0] in cart_code:

                dup_dict[d[0]]=(d[0],d[1])

       return dup_dict

    def post(self):

        args=self.reqparse.parse_args()

        cart_code=args['cart_code']

        if not cart_code:

            return {'error':'subid is Required!'}

        return {'cart_code':cart_code}

api.add_resource(PriceListAPI,'./price/',endpoint='stores')       #定義class類 對應的url

if __name__ == '__main__':

    app.run(debug=True,port=5091,houst='0.0.0.0')

 

經過 python app.py就能夠運行 

 

給API class添加裝飾器 ，在每次請求處理前先進行驗證 ，如經常使用的驗證 Headers中的Authorization字段

將app_key和app_secret融合 加密到組成 Authorization的值，來判斷用戶是否爲合法用戶

 

首先須要定義一個裝飾器函數

from functools import wraps 

def my_authenticate(func):

    @wraps(func)

    def wrapper(*args,**kwargs):

        authen = request.headers.get('Authorization')

        if not authen:
            return {"code":403,"message":"Authenticated Error","data":{"isEnabled":0}},403

        logger.warning('authen:%s'%(authen))

        encode_authen=authen.encode('utf-8')

        decode_authen = base64.b64decode(encode_authen)

        app_key,app_sign,timestamp = decode_authen.decode('utf-8').split(';')

        now = time.time()

        if now-float(timestamp)>2*60:

            return {"code":403,"message":"Authenticated expired","data":{"isEnabled":0}},403

        if app_key != APP_KEY:

            return {"code":403,"message":"Authenticate app_key error","data":{"isEnabled":0}},403

        md5_obj = hashlib.md5()

        content = app_key+APP_SECRET+timestamp

        md5_obj.update(content.encode('utf-8'))

        my_sign=md5_obj.hexdigest()

        if app_sign==my_sign:

            logger.info("authen ok!")

            return func(*args,**kwargs)

        return {"code":403,"message":"Authenticated Error","data":{"isEnabled":0}},403

    return wrapper

 

而後在class ProceListAPI(Resource)的開頭加上  decorators=[my_authenticate]

如：

class ProceListAPI(Resource):

    decorators=[my_authenticate]

    def __init__(self):

        self.reqparse=reqparse.RequestParser()

....

這樣就至關於給類ProceListAPI的每一個方法都加上了裝飾器 my_authenticate