当前位置:   article > 正文

Flask搭建api服务

flask部署api

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

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

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

9f93e13632d9b5a77f3ea76746b87714.png

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

configs.py

  1. HOST = '127.0.0.1'
  2. PORT = '5432'
  3. DATABASE = 'runoobdb'
  4. USERNAME = 'postgres'
  5. PASSWORD = '*****'
  6. # 配置主数据库
  7. DB_URI = "postgresql+psycopg2://{username}:{password}@{host}:{port}/{db}".format(username=USERNAME, password=PASSWORD,
  8. host=HOST, port=PORT, db=DATABASE)
  9. # SQLALCHEMY_DATABASE_URI = 'postgresql+psycopg2://postgres:*****@127.0.0.1:5432/runoobdb'
  10. # 连接其他数据库
  11. SQLALCHEMY_BINDS = {
  12. 'xxxdb': 'postgresql+psycopg2://postgres:123456@localhost:5432/lincms3',
  13. 'yyydb': 'postgresql+psycopg2://postgres:123456@localhost:5432/lincms4',
  14. 'zzzdb': 'sqlite:///users.db'
  15. }
  16. SQLALCHEMY_DATABASE_URI = DB_URI
  17. SQLALCHEMY_TRACK_MODIFICATIONS = False
  18. SQLALCHEMY_ECHO = True

第二步,在exts中定义全局db

exts.py

  1. from flask_sqlalchemy import SQLAlchemy
  2. db = SQLAlchemy()

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

flaskutils.py

  1. import decimal
  2. import numpy as np
  3. import json, datetime,configparser
  4. class DataEncoder(json.JSONEncoder):
  5. """数据转码类 """
  6. def default(self, obj):
  7. """针对无法转json的数据类型进行转码
  8. 目前支持的转码类型 1、将Numpy的intger,floating转为int和float
  9. 2、将Numpy的ndarray转为list
  10. 3、将np.datetime64转化为字符串前10位 4、将datetime.datetime转化为"%Y-%m-%d %H:%M:%S"
  11. 5、将datetime.date转化为"%Y-%m-%d"
  12. 6、将bytes转化为utf-8字符串
  13. 入参:
  14. obj: 数据对象
  15. 出参:
  16. 转化后的数据
  17. 异常:
  18. 无 """
  19. if isinstance(obj, np.integer):
  20. return int(obj)
  21. elif isinstance(obj, np.floating):
  22. return float(obj)
  23. elif isinstance(obj, np.ndarray):
  24. return obj.tolist()
  25. elif isinstance(obj, np.datetime64):
  26. return str(obj)[:10]
  27. elif isinstance(obj, datetime.datetime):
  28. return obj.strftime("%Y-%m-%d %H:%M:%S")
  29. elif isinstance(obj, datetime.date):
  30. return obj.strftime("%Y-%m-%d")
  31. elif isinstance(obj, decimal.Decimal):
  32. return float(obj)
  33. elif isinstance(obj, bytes):
  34. return str(obj, encoding='utf-8')
  35. else:
  36. return json.JSONEncoder.default(self, obj)
  37. def getsqlresultjson(db, sql,params={}):
  38. """根据db和sql语句,将结果集转换为json格式
  39. 根据db和sql语句,将结果集转换为json格式
  40. 第一步:根据cursor获取元数据,生成键值列表
  41. 第二步:遍历结果集,将键值列表和结果集组装成字典,加入列表
  42. 第三步:将列表通过DataEncoder进行转码
  43. 入参:
  44. db: 数据库实例.
  45. sql: 待运行的SQL语句
  46. 出参:
  47. Json格式:
  48. 举例: {'Serak': ('Rigel VII', 'Preparer'),
  49. 'Zim': ('Irk', 'Invader'),
  50. 'Lrrr': ('Omicron Persei 8', 'Emperor')}
  51. 异常:
  52. 无 """
  53. resultdict = []
  54. cursor = db.session.execute(sql,params=params).cursor
  55. resultproxy = db.session.execute(sql,params=params).fetchall()
  56. # 获取元数据
  57. colname = [i[0] for i in cursor.description]
  58. # 获取结果集,组成字典,加入列表
  59. for rowproxy in resultproxy:
  60. rowresult = dict(zip(colname, rowproxy))
  61. resultdict.append(rowresult)
  62. # 生成json格式
  63. jsonstr = json.dumps(resultdict, cls=DataEncoder)
  64. return jsonstr
  65. def parasecommaparamtolist(param):
  66. '''
  67. 处理in传递参数,in传递参数可适用于两种传递方式,逗号传递参数或参数传递
  68. 此处主要是处理,逗号传递参数,返回为list
  69. # http://127.0.0.1:5000/getresultbysqlgetparaminbylist?sqlid=sql10&begindate=2018&enddate=2020&kpicode=03010101
  70. # http://127.0.0.1:5000/getresultbysqlgetparaminbylist?sqlid=sql10&begindate=2018&enddate=2020&kpicode=03010101&kpicode=031111111
  71. # http://127.0.0.1:5000/getresultbysqlgetparaminbylist?sqlid=sql10&begindate=2018&enddate=2020
  72. # http://127.0.0.1:5000/getresultbysqlgetparaminbylist?sqlid=sql10&begindate=2018&enddate=2020&kpicode=03010101,222222222
  73. # http://127.0.0.1:5000/getresultbysqlgetparaminbylist?sqlid=sql10&begindate=2018&enddate=2020&kpicode=03010101&kpicode=03010101
  74. :param param:
  75. :return:
  76. 字符串列表 '''
  77. result = []
  78. for val in param.split(','):
  79. if val:
  80. result.append(val)
  81. return result

第四步,在app文件构建初始版本

app.py

  1. import configs
  2. from exts import db
  3. from flask import Flask
  4. from flaskutils import *
  5. from flask import request,jsonify
  6. app = Flask(__name__)
  7. # 加载配置文件
  8. app.config.from_object(configs)
  9. app.debug = True
  10. db.init_app(app)
  11. if __name__ == '__main__':
  12. print(app.url_map)
  13. app.run(host='0.0.0.0', port=8080)

第五步,在app文件中配置sql语句,原本想尝试一下mybis类型的配置文件,后来决定简化;主要包括三条sql,第一条不需要传参,第二条传递常规参数,第三条传递in参数,尤其是in参数,基本上网上找到的方法都不可靠,本文算是原创吧。

  1. sqldict={}
  2. sqldict['sql1'] = """select a.*
  3. from kpi_value a
  4. where a.kpicode in ('01010101','02010101','03010101')
  5. and a.datelevel='01'
  6. and a.regionlevel='02'
  7. """
  8. sqldict['sql2'] = """select a.*
  9. from kpi_value a
  10. where a.kpicode in ('01010101','02010101','03010101')
  11. and a.datelevel='01'
  12. and a.regionlevel='02'
  13. and a.datecode>=:begindate and a.datecode<=:enddate
  14. """
  15. sqldict['sql3'] = """select a.*
  16. from kpi_value a
  17. and a.datelevel='01'
  18. and a.regionlevel='02'
  19. and a.datecode>=:begindate and a.datecode<=:enddate
  20. and a.kpicode in :kpicode
  21. """

1、构造第一个最简单sql返回接口,不需要传递sql参数,但需要传递sqlid参数

  1. @app.route('/getresultbysql', methods=['GET', 'POST'])
  2. def index1():
  3. sqlid = request.args.get('sqlid')
  4. sqltext=sqldict[sqlid]
  5. jsonstr = getsqlresultjson(db,sqltext)
  6. return jsonstr, 200, {"Content-Type": "application/json"}

2、构造一个sql内部传参的接口,通过字典参数方式

  1. @app.route('/getresultbysqlparam', methods=['GET', 'POST'])
  2. def index2():
  3. sqlid = request.args.get('sqlid')
  4. sqltext=sqldict[sqlid]
  5. params = {"begindate": '2017',"enddate":'2019'}
  6. jsonstr = getsqlresultjson(db,sqltext,params)
  7. return jsonstr, 200, {"Content-Type": "application/json"}

3、通过url进行sql参数的传递。

  1. @app.route('/getresultbysqlgetparam', methods=['GET', 'POST'])
  2. def index3():
  3. sqlid = request.args.get('sqlid')
  4. begindate = request.args.get('begindate')
  5. enddate = request.args.get('enddate')
  6. sqltext=sqldict[sqlid]
  7. params = {"begindate": begindate,"enddate":enddate}
  8. jsonstr = getsqlresultjson(db,sqltext,params)
  9. return jsonstr, 200, {"Content-Type": "application/json"}

4、通过url进行sql参数的传递,不过不传递in参数,而是在路由函数汇总内部指定in参数

  1. @app.route('/getresultbysqlgetparamin', methods=['GET', 'POST'])
  2. def index4():
  3. sqlid = request.args.get('sqlid')
  4. sqlid='sql3'
  5. begindate = request.args.get('begindate')
  6. enddate = request.args.get('enddate')
  7. sqltext=sqldict[sqlid]
  8. incond = ['01010101', '03010101']
  9. params = {"begindate": begindate,"enddate":enddate,'kpicode':tuple(incond)}
  10. jsonstr = getsqlresultjson(db,sqltext,params)
  11. return jsonstr, 200, {"Content-Type": "application/json"}

5、通过url进行in参数和普通参数的传递,这里可以支持两种方式,一种是&aa=xxx&aa=yyy,一种是aa=xxx,yyy。

  1. @app.route('/getresultbysqlgetparaminbylist', methods=['GET', 'POST'])
  2. def index5():
  3. sqlid = request.args.get('sqlid')
  4. sqlid='sql3'
  5. begindate = request.args.get('begindate')
  6. enddate = request.args.get('enddate')
  7. incond=request.args.getlist('kpicode')
  8. if len(incond) == 1 and ',' in incond[0]:
  9. incond = parasecommaparamtolist(incond[0])
  10. sqltext=sqldict[sqlid]
  11. params = {"begindate": begindate,"enddate":enddate,'kpicode':tuple(incond)}
  12. jsonstr = getsqlresultjson(db,sqltext,params)
  13. return jsonstr, 200, {"Content-Type": "application/json"}

6、标准化接口响应返回结果。

  1. @app.route('/getresultbysqlgetparaminbylistresponse', methods=['GET', 'POST'])
  2. def index6():
  3. retinfo={}
  4. errorflag=False
  5. retinfo['returncode'] = 200
  6. retinfo['returndata'] = ''
  7. retinfo['returninfo'] = '处理成果'
  8. sqlid = request.args.get('sqlid')
  9. begindate = request.args.get('begindate')
  10. enddate = request.args.get('enddate')
  11. incond = request.args.getlist('kpicode')
  12. if len(incond) == 1 and ',' in incond[0]:
  13. incond = parasecommaparamtolist(incond[0])
  14. if not incond:
  15. retinfo['returninfo']=retinfo['returninfo'] +'未传入KPI编码'
  16. errorflag=True
  17. if not begindate:
  18. retinfo['returninfo'] = retinfo['returninfo'] + '未传入开始时间'
  19. errorflag=True
  20. if not enddate:
  21. retinfo['returninfo'] = retinfo['returninfo'] + '未传入结束时间'
  22. errorflag=True
  23. if begindate>enddate:
  24. retinfo['returninfo'] = retinfo['returninfo'] + '开始时间大于结束时间'
  25. errorflag=True
  26. if errorflag==True:
  27. retinfo['returncode'] = 400
  28. response = jsonify(retinfo)
  29. response.status_code = 400
  30. return response
  31. sqltext = sqldict[sqlid]
  32. params = {"begindate": begindate, "enddate": enddate, 'kpicode': tuple(incond)}
  33. jsonstr = getsqlresultjson(db, sqltext, params)
  34. retinfo['returndata'] = jsonstr
  35. response = jsonify(retinfo)
  36. response.status_code = 200
  37. return response

最后,谢谢关注,谢谢支持!

1c832f5feafb52fb69ec196a77741e70.png

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/知新_RL/article/detail/791542
推荐阅读
相关标签
  

闽ICP备14008679号