当前位置:网站首页>Flash build API service
Flash build API service
2022-07-07 17:07:00 【Python and big data analysis】
Flask It's a use Python Written lightweight Web Application framework , Very suitable for personal development , We make an interface here .
For the convenience of debugging , This article USES the get The interface way .get The interface is very simple , There is no need to upload any data , Add a... After the path get The method can be used , The returned string is .
This article is just Flask Preliminary documentation of the developed interface , From the simplest interface development to the slightly more complex interface , If there is time in the future , Will gradually improve , Include token authentication 、 Cross domain authentication 、 Blueprint application 、 Log management, etc .
First step , First, in the configs Configure data source in
configs.py
HOST = '127.0.0.1'
PORT = '5432'
DATABASE = 'runoobdb'
USERNAME = 'postgres'
PASSWORD = '*****'
# Configure the main database
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'
# Connect to other databases
SQLALCHEMY_BINDS = {
'xxxdb': 'postgresql+psycopg2://postgres:[email protected]:5432/lincms3',
'yyydb': 'postgresql+psycopg2://postgres:[email protected]:5432/lincms4',
'zzzdb': 'sqlite:///users.db'
}
SQLALCHEMY_DATABASE_URI = DB_URI
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO = True
The second step , stay exts Define global in db
exts.py
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
The third step , Constructed a flaskutils, Define some public classes to which the interface applies , For example, data transcoding , Convert the data set to json, analysis url Comma parameter, etc , The functions will be expanded on this basis in the future .
flaskutils.py
import decimal
import numpy as np
import json, datetime,configparser
class DataEncoder(json.JSONEncoder):
""" Data transcoding class """
def default(self, obj):
""" For unable to transfer json Transcoding the data type of
Currently supported transcoding types 1、 take Numpy Of intger,floating To int and float
2、 take Numpy Of ndarray To list
3、 take np.datetime64 Before converting to string 10 position 4、 take datetime.datetime Turn into "%Y-%m-%d %H:%M:%S"
5、 take datetime.date Turn into "%Y-%m-%d"
6、 take bytes Turn into utf-8 character string
Enter the reference :
obj: Data objects
The ginseng :
Transformed data
abnormal :
nothing """
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={}):
""" according to db and sql sentence , Convert result set to json Format
according to db and sql sentence , Convert result set to json Format
First step : according to cursor Fetch metadata , Generate key value list
The second step : Traversal result set , Assemble the key value list and result set into a dictionary , Join list
The third step : Pass the list through DataEncoder Transcoding
Enter the reference :
db: Database instance .
sql: To be run SQL sentence
The ginseng :
Json Format :
give an example : {'Serak': ('Rigel VII', 'Preparer'),
'Zim': ('Irk', 'Invader'),
'Lrrr': ('Omicron Persei 8', 'Emperor')}
abnormal :
nothing """
resultdict = []
cursor = db.session.execute(sql,params=params).cursor
resultproxy = db.session.execute(sql,params=params).fetchall()
# Fetch metadata
colname = [i[0] for i in cursor.description]
# Get the result set , Make up a dictionary , Join list
for rowproxy in resultproxy:
rowresult = dict(zip(colname, rowproxy))
resultdict.append(rowresult)
# Generate json Format
jsonstr = json.dumps(resultdict, cls=DataEncoder)
return jsonstr
def parasecommaparamtolist(param):
'''
Handle in Pass parameters ,in The transfer parameters can be applied to two transfer methods , Comma passing parameter or parameter passing
Here we mainly deal with , Pass arguments with commas , Return to 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:
String list '''
result = []
for val in param.split(','):
if val:
result.append(val)
return result
Step four , stay app File build initial version
app.py
import configs
from exts import db
from flask import Flask
from flaskutils import *
from flask import request,jsonify
app = Flask(__name__)
# Load profile
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)
Step five , stay app Configuration in file sql sentence , I wanted to try mybis Type of profile , Later decided to simplify ; It mainly includes three items sql, Article 1 no reference is required , Second, pass the general parameters , Article 3 transmission in Parameters , In especial in Parameters , Basically, the methods found on the Internet are not reliable , This article is original .
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、 Constructing the first one is the simplest sql Return Interface , There is no need to pass on sql Parameters , But you need to pass sqlid Parameters
@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、 Construct a sql Interface for internal parameter transfer , By means of dictionary parameters
@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、 adopt url Conduct sql Parameter passing .
@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、 adopt url Conduct sql Parameter passing , But don't pass in Parameters , Instead, it is specified in the routing function summary in Parameters
@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、 adopt url Conduct in The transfer of parameters and common parameters , There are two ways to support this , One is &aa=xxx&aa=yyy, One is 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、 The standardized interface response returns the result .
@app.route('/getresultbysqlgetparaminbylistresponse', methods=['GET', 'POST'])
def index6():
retinfo={}
errorflag=False
retinfo['returncode'] = 200
retinfo['returndata'] = ''
retinfo['returninfo'] = ' Processing results '
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'] +' No incoming KPI code '
errorflag=True
if not begindate:
retinfo['returninfo'] = retinfo['returninfo'] + ' No start time was passed in '
errorflag=True
if not enddate:
retinfo['returninfo'] = retinfo['returninfo'] + ' End time not passed in '
errorflag=True
if begindate>enddate:
retinfo['returninfo'] = retinfo['returninfo'] + ' The start time is greater than the end time '
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
Last , Thank you for your attention , Thank you for your support !
边栏推荐
- LeetCode 1981. Minimize the difference between the target value and the selected element one question per day
- 蓝桥杯 决赛 异或变换 100分
- 最新Android高级面试题汇总,Android面试题及答案
- 如何选择合适的自动化测试工具?
- 模块六
- Talk about the realization of authority control and transaction record function of SAP system
- QT中自定义控件的创建到封装到工具栏过程(二):自定义控件封装到工具栏
- 谈谈 SAP 系统的权限管控和事务记录功能的实现
- 科普达人丨一文弄懂什么是云计算?
- The latest interview experience of Android manufacturers in 2022, Android view+handler+binder
猜你喜欢
谈谈 SAP 系统的权限管控和事务记录功能的实现
How to add aplayer music player in blog
Skimage learning (1)
Seaborn数据可视化
QT 图片背景色像素处理法
Pychart ide Download
[Seaborn] combination chart: facetgrid, jointgrid, pairgrid
低代码(lowcode)帮助运输公司增强供应链管理的4种方式
Master this promotion path and share interview materials
整理几个重要的Android知识,高级Android开发面试题
随机推荐
【视频/音频数据处理】上海道宁为您带来Elecard下载、试用、教程
Master this promotion path and share interview materials
LocalStorage和SessionStorage
LeetCode 403. 青蛙过河 每日一题
Skimage learning (2) -- RGB to grayscale, RGB to HSV, histogram matching
LeetCode 312. Poke balloon daily
dapp丨defi丨nft丨lp单双币流动性挖矿系统开发详细说明及源码
《产品经理必读:五种经典的创新思维模型》的读后感
QT视频传输
Shallow understanding Net core routing
LeetCode 1774. 最接近目标价格的甜点成本 每日一题
QML beginner
Flask搭建api服务
Module VI
LeetCode 403. Frog crossing the river daily
电脑无法加域,ping域名显示为公网IP,这是什么问题?怎么解决?
从DevOps到MLOps:IT工具怎样向AI工具进化?
Sator推出Web3游戏“Satorspace” ,并上线Huobi
国内首创!Todesk将RTC技术融入远程桌面,画质更清晰操作更流畅
Binary search tree (basic operation)