一、安装必要的库
首先，确保已经安装了以下库：
- 有网模式
pip install flask
pip install gevent
pi install cx_Oracle
- 离线模式：
下载地址：https://pypi.org/simple/flask/
# a. Flask
Werkzeug-1.0.1-py2.py3-none-any.whl
Jinja2-2.11.3-py2.py3-none-any.whl
--|MarkupSafe-1.1.1-cp38-cp38-win_amd64.whl
itsdangerous-1.1.0-py2.py3-none-any.whl
click-7.1.2-py2.py3-none-any.whl
Flask-1.1.4-py2.py3-none-any.whl# b. gevent
# 注意安装最新版本，否则会出现问题
zope.event-4.6-py2.py3-none-any.whl
zope.interface-7.1.1-cp38-cp38-win_amd64.whl
greenlet-3.1.1-cp38-cp38-win_amd64.whl
cffi-1.16.0-cp38-cp38-win_amd64.whl
-|pycparser-2.20-py2.py3-none-any.whl
gevent-24.2.1-cp38-cp38-win_amd64.whl
二、代码实现
- reply_oracle.py（主应用文件）
from flask import Flask
from datetime import datetime
from gevent import pywsgi
import cx_Oracle
import jsonapp = Flask(__name__)# ip:1521/etltdbdb_host = "ip"
db_port = "1521"
db_instant = "etltdb"
user = "dc_ctl"
password = "******"sql = """ select t.db_name,t.db_type,t.db_user,t.db_passwd,t.db_host,t.db_port,t.db_instantfrom dc_ctl.ctl_db_info t
"""@app.route('/oracle')
def get_current_time():# 创建数据库连接dsn = cx_Oracle.makedsn(db_host, db_port, db_instant)con = cx_Oracle.connect(user, password, dsn)# 创建游标cursor = con.cursor()# 执行查询SQLcursor.execute(sql)# 获取结果集rows = cursor.fetchall()# 将结果集转换成JSON格式result = []for row in rows:result.append({'db_name': row[0],'db_type': row[1],'db_passwd': row[2],'db_host': row[3],'db_port': row[4],'db_instant': row[5]})json_result = json.dumps(result)return json_resultif __name__ == '__main__':print(datetime.now().strftime('%Y-%m-%d %H:%M:%S'))server = pywsgi.WSGIServer(('0.0.0.0', 5001), app)server.serve_forever()
三、测试
- 测试方式1-代码
import requestsdef app():response = requests.get("http://ip:5001/oracle")print(response.json())if __name__ == '__main__':app()
- 测试方式2-界面
http://ip:5001/oracle
