2021-04-20

flask_oracle

###1.service 文件

import cx_Oracle
import pandas as pd
con=cx_Oracle.connect("scott/123@localhost:1521/orcl",encoding='utf-8')

# 一.service 文件
# 1.查询员工信息（无参数时返回全量信息）
def SearchEmpInfor(n_deptno):
    v_where = ' and 1=1 '
    if n_deptno != "":
        v_where = v_where + f''' and instr({n_deptno},deptno) >0 '''        
    try:
        code, msg = 1, "查询成功"
        sql= f''' SELECT e.empno, e.ename, e.job, e.mgr, e.hiredate, e.sal, e.comm, e.deptno
                      FROM emp e
                     WHERE 1=1 {v_where} '''
        df = pd.read_sql(sql,con)
    except Exception as e:
        code, msg, df = -1, "数据查询错误:{}".formate(e), ""
        print(msg)
    finally:
        if 'con' in dir():
            con.close()
    return code, msg, df


# 2.查询员工所在部门的信息（无参数时返回全量信息）
def SearchDeptInfor(n_empno):
    v_where = ' and 1=1 '
    if n_empno != "":
        v_where = v_where + f''' and instr({n_empno},empno) >0 '''        
    try:
        code, msg = 1, "查询成功"
        sql= f'''  SELECT e.ename, d.dname, d.loc
                       FROM emp e
                       LEFT JOIN dept d
                         ON e.deptno = d.deptno
                      WHERE 1 = 1 {v_where} '''
        df = pd.read_sql(sql,con)
    except Exception as e:
        code, msg, df = -1, "数据查询错误:{}".formate(e), ""
        print(msg)
    finally:
        if 'con' in dir():
            con.close()
    return code, msg, df

if __name__ == '__main__':
    n_deptno = 10
    n_empno = '7369'
    a = SearchEmpInfor(n_deptno)[2]
    b = SearchDeptInfor(n_empno)[2]
    print(a)
    print(b)

# 二.service_total文件(用于处理，整合service)
def SearchTatol(n_deptno, n_empno):
    try:
        code, msg = 1, "查询成功"
        df = SearchEmpInfor(n_deptno)[2]
        EmpInfor = df[["EMPNO","ENAME","SAL","DEPTNO"]].rename(columns={"EMPNO":"员工编号","ENAME":"员工姓名","DEPTNO":"部门编号"})
        
        df = SearchDeptInfor(n_empno)[2]
        DeptInfor = df[["ENAME","DNAME","LOC"]].rename(columns={"ENAME":"员工姓名","DNAME":"部门名称","LOC":"部门地址",})
        
        result = {"EmpInfor":EmpInfor.to_dict(orient= "records"),
                 "DeptInfor":DeptInfor.to_dict(orient= "records")
                 }
    except Exception as e:
        code, msg, result = -1, "数据查询错误:{}".format(e), ""
        print(msg)
    return code, msg, result

if __name__ == '__main__':
    n_deptno = 10
    n_empno = '7369'
    a = SearchTatol(n_deptno, n_empno)
    print(a)

# 三.views 文件
from flask import Blueprint,make_response,request,jsonify,json,Flask,url_for,request,render_template,redirect,session
from flask_cors import CORS

bp = Blueprint("invest_report_bp",__name__)
CORS(bp,resources = r"/*")

# 1.查询数据汇总
@bp.route('/SearchTatol_view', methods = ['post','get'])
def SearchTatol_view():
    try:
        # 获取参数(json入参)
        n_deptno = ""
        n_empno = ""
        params = json.loads(request.get_data())
        if "n_deptno" in params:
            n_deptno = params['n_deptno']
        if "n_deptno" in params:
            n_empno = params['n_empno']
        
        # 带入参数获取结果集
        code, msg, df = SearchTatol(n_deptno, n_empno)
        data = json.loads(json.dumps(df))
        return make_response(jsonify(code=code,msg=msg,data=data))
    except Exception as e:
        # logger.error("...",traceback.format_exc())
        return make_response(jsonify(code=-1,msg=str(e),data=""))
    

# 四.运行run
#import 
def init_app(app):
    """
    注册接口
    @param app:
    """
    app.register_blueprint(bp, url='/data/service_tatol')
    
def create_app():
    #import data
    app = Flask(__name__)
    CORS(app,resources=r"/*")
    init_app(app)
    return app

app = create_app()
@app.errorhandler(Exception)
def error_500():
    respose = dict(status=0,msg="500 error")
    return jsonify(respose), 400


if __name__ == '__main__':
    app.run('0.0.0.0',8085)

最后再postman post接口（入参，获取）