赞
踩
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

最后,谢谢关注,谢谢支持!
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。