Flask搭建api服務(wù)的實現(xiàn)步驟

更新時間：2022年06月13日 09:24:06 作者：python與大數(shù)據(jù)分析

本文主要介紹了Flask搭建api服務(wù)的實現(xiàn)步驟，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

Flask是一個使用 Python 編寫的輕量級 Web 應(yīng)用框架，很適合個人開發(fā)，我們在此處做一個接口。

為方便調(diào)試，本文使用get接口方式。get接口十分簡單，不需要上傳任何數(shù)據(jù)，在路徑后面添加一個get方法就可以用，返回的是字符串。

本文只是Flask開發(fā)的接口的初步文檔，從最簡單的接口開發(fā)到稍微復(fù)雜一些的接口，后續(xù)如有時間，會逐步完善，包括token鑒權(quán)、跨域認證、藍圖應(yīng)用、日志管理等等。

第一步，首先在configs中配置數(shù)據(jù)源

configs.py

HOST = '127.0.0.1'
PORT = '5432'
DATABASE = 'runoobdb'
USERNAME = 'postgres'
PASSWORD = '*****'
# 配置主數(shù)據(jù)庫
DB_URI = "postgresql+psycopg2://{username}:{password}@{host}:{port}/{db}".format(username=USERNAME, password=PASSWORD,
                                                                                 host=HOST, port=PORT, db=DATABASE)
# SQLALCHEMY_DATABASE_URI = 'postgresql+psycopg2://postgres:*****@127.0.0.1:5432/runoobdb'
# 連接其他數(shù)據(jù)庫
SQLALCHEMY_BINDS = {
    'xxxdb': 'postgresql+psycopg2://postgres:123456@localhost:5432/lincms3',
    'yyydb': 'postgresql+psycopg2://postgres:123456@localhost:5432/lincms4',
    'zzzdb': 'sqlite:///users.db'
}
SQLALCHEMY_DATABASE_URI = DB_URI
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO = True

第二步，在exts中定義全局db

exts.py

from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()

第三步，構(gòu)造了一個flaskutils，在這里定義一些接口應(yīng)用到的公共類，比如數(shù)據(jù)轉(zhuǎn)碼，將數(shù)據(jù)集轉(zhuǎn)換為json，解析url逗號參數(shù)等等，后續(xù)將在此基礎(chǔ)上拓展功能。

flaskutils.py

import decimal
 
import numpy as np
import json, datetime,configparser
 
 
class DataEncoder(json.JSONEncoder):
    """數(shù)據(jù)轉(zhuǎn)碼類    """
    def default(self, obj):
        """針對無法轉(zhuǎn)json的數(shù)據(jù)類型進行轉(zhuǎn)碼
        目前支持的轉(zhuǎn)碼類型        1、將Numpy的intger,floating轉(zhuǎn)為int和float
        2、將Numpy的ndarray轉(zhuǎn)為list
        3、將np.datetime64轉(zhuǎn)化為字符串前10位        4、將datetime.datetime轉(zhuǎn)化為"%Y-%m-%d %H:%M:%S"
        5、將datetime.date轉(zhuǎn)化為"%Y-%m-%d"
        6、將bytes轉(zhuǎn)化為utf-8字符串
        入?yún)?
            obj: 數(shù)據(jù)對象
        出參:
            轉(zhuǎn)化后的數(shù)據(jù)
        異常:
            無        """
        if isinstance(obj, np.integer):
            return int(obj)
        elif isinstance(obj, np.floating):
            return float(obj)
        elif isinstance(obj, np.ndarray):
            return obj.tolist()
        elif isinstance(obj, np.datetime64):
            return str(obj)[:10]
        elif isinstance(obj, datetime.datetime):
            return obj.strftime("%Y-%m-%d %H:%M:%S")
        elif isinstance(obj, datetime.date):
            return obj.strftime("%Y-%m-%d")
        elif isinstance(obj, decimal.Decimal):
            return float(obj)
        elif isinstance(obj, bytes):
            return str(obj, encoding='utf-8')
        else:
            return json.JSONEncoder.default(self, obj)
 
 
def getsqlresultjson(db, sql,params={}):
    """根據(jù)db和sql語句，將結(jié)果集轉(zhuǎn)換為json格式
    根據(jù)db和sql語句，將結(jié)果集轉(zhuǎn)換為json格式
    第一步：根據(jù)cursor獲取元數(shù)據(jù)，生成鍵值列表
    第二步：遍歷結(jié)果集，將鍵值列表和結(jié)果集組裝成字典，加入列表
    第三步：將列表通過DataEncoder進行轉(zhuǎn)碼
    入?yún)?
        db: 數(shù)據(jù)庫實例.
        sql: 待運行的SQL語句
    出參:
        Json格式:
        舉例：        {'Serak': ('Rigel VII', 'Preparer'),
          'Zim': ('Irk', 'Invader'),
         'Lrrr': ('Omicron Persei 8', 'Emperor')}
    異常:
        無    """
    resultdict = []
 
    cursor = db.session.execute(sql,params=params).cursor
    resultproxy = db.session.execute(sql,params=params).fetchall()
 
    # 獲取元數(shù)據(jù)
    colname = [i[0] for i in cursor.description]
    # 獲取結(jié)果集，組成字典，加入列表
    for rowproxy in resultproxy:
        rowresult = dict(zip(colname, rowproxy))
        resultdict.append(rowresult)
 
    # 生成json格式
    jsonstr = json.dumps(resultdict, cls=DataEncoder)
    return jsonstr
 
 
def parasecommaparamtolist(param):
    '''
    處理in傳遞參數(shù)，in傳遞參數(shù)可適用于兩種傳遞方式，逗號傳遞參數(shù)或參數(shù)傳遞
    此處主要是處理,逗號傳遞參數(shù)，返回為list
    # http://127.0.0.1:5000/getresultbysqlgetparaminbylist?sqlid=sql10&begindate=2018&enddate=2020&kpicode=03010101
    # http://127.0.0.1:5000/getresultbysqlgetparaminbylist?sqlid=sql10&begindate=2018&enddate=2020&kpicode=03010101&kpicode=031111111
    # http://127.0.0.1:5000/getresultbysqlgetparaminbylist?sqlid=sql10&begindate=2018&enddate=2020
    # http://127.0.0.1:5000/getresultbysqlgetparaminbylist?sqlid=sql10&begindate=2018&enddate=2020&kpicode=03010101,222222222
    # http://127.0.0.1:5000/getresultbysqlgetparaminbylist?sqlid=sql10&begindate=2018&enddate=2020&kpicode=03010101&kpicode=03010101
    :param param:
    :return:
        字符串列表    '''
    result = []
    for val in param.split(','):
        if val:
            result.append(val)
    return result

第四步，在app文件構(gòu)建初始版本

app.py

import configs
from exts import db
from flask import Flask
from flaskutils import *
from flask import request,jsonify
 
app = Flask(__name__)
 
# 加載配置文件
app.config.from_object(configs)
app.debug = True
 
db.init_app(app)
 
if __name__ == '__main__':
    print(app.url_map)
    app.run(host='0.0.0.0', port=8080)

第五步，在app文件中配置sql語句，原本想嘗試一下mybis類型的配置文件，后來決定簡化；主要包括三條sql，第一條不需要傳參，第二條傳遞常規(guī)參數(shù)，第三條傳遞in參數(shù)，尤其是in參數(shù)，基本上網(wǎng)上找到的方法都不可靠，本文算是原創(chuàng)吧。

sqldict={}
sqldict['sql1'] = """select a.*
from kpi_value a
where a.kpicode in ('01010101','02010101','03010101')
and a.datelevel='01'
and a.regionlevel='02'
"""
sqldict['sql2'] = """select a.*
from kpi_value a
where a.kpicode in ('01010101','02010101','03010101')
and a.datelevel='01'
and a.regionlevel='02'
and a.datecode>=:begindate and a.datecode<=:enddate
"""
sqldict['sql3'] = """select a.*
from kpi_value a
and a.datelevel='01'
and a.regionlevel='02'
and a.datecode>=:begindate and a.datecode<=:enddate
and a.kpicode in :kpicode
"""

1、構(gòu)造第一個最簡單sql返回接口，不需要傳遞sql參數(shù)，但需要傳遞sqlid參數(shù)

@app.route('/getresultbysql', methods=['GET', 'POST'])
def index1():
    sqlid = request.args.get('sqlid')
    sqltext=sqldict[sqlid]
    jsonstr = getsqlresultjson(db,sqltext)
    return jsonstr, 200, {"Content-Type": "application/json"}

2、構(gòu)造一個sql內(nèi)部傳參的接口，通過字典參數(shù)方式

@app.route('/getresultbysqlparam', methods=['GET', 'POST'])
def index2():
    sqlid = request.args.get('sqlid')
    sqltext=sqldict[sqlid]
    params = {"begindate": '2017',"enddate":'2019'}
    jsonstr = getsqlresultjson(db,sqltext,params)
    return jsonstr, 200, {"Content-Type": "application/json"}

3、通過url進行sql參數(shù)的傳遞。

@app.route('/getresultbysqlgetparam', methods=['GET', 'POST'])
def index3():
    sqlid = request.args.get('sqlid')
    begindate = request.args.get('begindate')
    enddate = request.args.get('enddate')
    sqltext=sqldict[sqlid]
    params = {"begindate": begindate,"enddate":enddate}
    jsonstr = getsqlresultjson(db,sqltext,params)
    return jsonstr, 200, {"Content-Type": "application/json"}

4、通過url進行sql參數(shù)的傳遞，不過不傳遞in參數(shù)，而是在路由函數(shù)匯總內(nèi)部指定in參數(shù)

@app.route('/getresultbysqlgetparamin', methods=['GET', 'POST'])
def index4():
    sqlid = request.args.get('sqlid')
    sqlid='sql3'
    begindate = request.args.get('begindate')
    enddate = request.args.get('enddate')
    sqltext=sqldict[sqlid]
    incond = ['01010101',  '03010101']
    params = {"begindate": begindate,"enddate":enddate,'kpicode':tuple(incond)}
    jsonstr = getsqlresultjson(db,sqltext,params)
    return jsonstr, 200, {"Content-Type": "application/json"}

5、通過url進行in參數(shù)和普通參數(shù)的傳遞，這里可以支持兩種方式，一種是&aa=xxx&aa=yyy，一種是aa=xxx,yyy。

@app.route('/getresultbysqlgetparaminbylist', methods=['GET', 'POST'])
def index5():
    sqlid = request.args.get('sqlid')
    sqlid='sql3'
    begindate = request.args.get('begindate')
    enddate = request.args.get('enddate')
    incond=request.args.getlist('kpicode')
 
    if len(incond) == 1 and ',' in incond[0]:
        incond = parasecommaparamtolist(incond[0])
 
    sqltext=sqldict[sqlid]
    params = {"begindate": begindate,"enddate":enddate,'kpicode':tuple(incond)}
    jsonstr = getsqlresultjson(db,sqltext,params)
 
    return jsonstr, 200, {"Content-Type": "application/json"}

6、標準化接口響應(yīng)返回結(jié)果。

@app.route('/getresultbysqlgetparaminbylistresponse', methods=['GET', 'POST'])
def index6():
 
    retinfo={}
    errorflag=False
    retinfo['returncode'] = 200
    retinfo['returndata'] = ''
    retinfo['returninfo'] = '處理成果'
    sqlid = request.args.get('sqlid')
    begindate = request.args.get('begindate')
    enddate = request.args.get('enddate')
    incond = request.args.getlist('kpicode')
 
    if len(incond) == 1 and ',' in incond[0]:
        incond = parasecommaparamtolist(incond[0])
    if not incond:
        retinfo['returninfo']=retinfo['returninfo'] +'未傳入KPI編碼'
        errorflag=True
    if not begindate:
        retinfo['returninfo'] = retinfo['returninfo'] + '未傳入開始時間'
        errorflag=True
    if not enddate:
        retinfo['returninfo'] = retinfo['returninfo'] + '未傳入結(jié)束時間'
        errorflag=True
    if begindate>enddate:
        retinfo['returninfo'] = retinfo['returninfo'] + '開始時間大于結(jié)束時間'
        errorflag=True
    if errorflag==True:
        retinfo['returncode'] = 400
        response = jsonify(retinfo)
        response.status_code = 400
        return response
 
    sqltext = sqldict[sqlid]
    params = {"begindate": begindate, "enddate": enddate, 'kpicode': tuple(incond)}
    jsonstr = getsqlresultjson(db, sqltext, params)
    retinfo['returndata']  = jsonstr
    response = jsonify(retinfo)
    response.status_code = 200
    return response

到此這篇關(guān)于Flask搭建api服務(wù)的實現(xiàn)步驟的文章就介紹到這了,更多相關(guān)Flask搭建api服務(wù)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: