Flask搭建api服务的实现步骤
短信预约 -IT技能 免费直播动态提醒
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服务内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！
免责声明：
① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的，并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据，供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。
② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341
