Flask搭建api服务的实现步骤

发布时间：2024-07-09 08:01

Flask是一个使用 Python 编写的轻量级 Web 应用框架，很适合个人开发，我们在此处做一个接口。

为方便调试，本文使用get接口方式。get接口十分简单，不需要上传任何数据，在路径后面添加一个get方法就可以用，返回的是字符串。

本文只是Flask开发的接口的初步文档，从最简单的接口开发到稍微复杂一些的接口，后续如有时间，会逐步完善，包括token鉴权、跨域认证、蓝图应用、日志管理等等。

第一步，首先在configs中配置数据源

configs.py

HOST = \'127.0.0.1\'
PORT = \'5432\'
DATABASE = \'runoobdb\'
USERNAME = \'postgres\'
PASSWORD = \'*****\'
# 配置主数据库
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\'
# 连接其他数据库
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()

第三步，构造了一个flaskutils，在这里定义一些接口应用到的公共类，比如数据转码，将数据集转换为json，解析url逗号参数等等，后续将在此基础上拓展功能。

flaskutils.py

import decimal
 
import numpy as np
import json, datetime,configparser
 
 
class DataEncoder(json.JSONEncoder):
    \"\"\"数据转码类    \"\"\"
    def default(self, obj):
        \"\"\"针对无法转json的数据类型进行转码
        目前支持的转码类型        1、将Numpy的intger,floating转为int和float
        2、将Numpy的ndarray转为list
        3、将np.datetime64转化为字符串前10位        4、将datetime.datetime转化为\"%Y-%m-%d %H:%M:%S\"
        5、将datetime.date转化为\"%Y-%m-%d\"
        6、将bytes转化为utf-8字符串
        入参:
            obj: 数据对象
        出参:
            转化后的数据
        异常:
            无        \"\"\"
        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={}):
    \"\"\"根据db和sql语句，将结果集转换为json格式
    根据db和sql语句，将结果集转换为json格式
    第一步：根据cursor获取元数据，生成键值列表
    第二步：遍历结果集，将键值列表和结果集组装成字典，加入列表
    第三步：将列表通过DataEncoder进行转码
    入参:
        db: 数据库实例.
        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()
 
    # 获取元数据
    colname = [i[0] for i in cursor.description]
    # 获取结果集，组成字典，加入列表
    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传递参数，in传递参数可适用于两种传递方式，逗号传递参数或参数传递
    此处主要是处理,逗号传递参数，返回为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文件构建初始版本

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，第一条不需要传参，第二条传递常规参数，第三条传递in参数，尤其是in参数，基本上网上找到的方法都不可靠，本文算是原创吧。

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、构造第一个最简单sql返回接口，不需要传递sql参数，但需要传递sqlid参数

@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、构造一个sql内部传参的接口，通过字典参数方式

@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参数的传递。

@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参数的传递，不过不传递in参数，而是在路由函数汇总内部指定in参数

@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参数和普通参数的传递，这里可以支持两种方式，一种是&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、标准化接口响应返回结果。

@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\'] + \'未传入结束时间\'
        errorflag=True
    if begindate>enddate:
        retinfo[\'returninfo\'] = retinfo[\'returninfo\'] + \'开始时间大于结束时间\'
        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

到此这篇关于Flask搭建api服务的实现步骤的文章就介绍到这了,更多相关Flask搭建api服务内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

Flask搭建api服务的实现步骤

相关推荐