当前位置:网站首页>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 !
边栏推荐
猜你喜欢
最新高频Android面试题目分享,带你一起探究Android事件分发机制
Lowcode: four ways to help transportation companies enhance supply chain management
node:504报错
Sator推出Web3游戏“Satorspace” ,并上线Huobi
面向接口编程
掌握这套精编Android高级面试题解析,oppoAndroid面试题
科普达人丨一文弄懂什么是云计算?
[image sensor] correlated double sampling CDs
最新Android高级面试题汇总,Android面试题及答案
Pisa-Proxy SQL 解析之 Lex & Yacc
随机推荐
[image sensor] correlated double sampling CDs
skimage学习(1)
Lowcode: four ways to help transportation companies enhance supply chain management
LeetCode 1186. Delete once to get the sub array maximum and daily question
【饭谈】那些看似为公司着想,实际却很自私的故事 (一:造轮子)
Skimage learning (2) -- RGB to grayscale, RGB to HSV, histogram matching
面向接口编程
编程模式-表驱动编程
直接上干货,100%好评
Ray and OBB intersection detection
蓝桥杯 决赛 异或变换 100分
【饭谈】如何设计好一款测试平台?
time标准库
正在准备面试,分享面经
MRS离线数据分析:通过Flink作业处理OBS数据
LeetCode 152. 乘积最大子数组 每日一题
Sator推出Web3游戏“Satorspace” ,并上线Huobi
Proxmox VE重装后,如何无损挂载原有的数据盘?
自定义View必备知识,Android研发岗必问30+道高级面试题
Flask搭建api服务-生成API文档