cx_Oracle 操作oracle數據庫

cx_Oracle 操作oracle數據庫

參考 https://www.jianshu.com/p/78e6640bf4b0 安裝配置cx_Oracle模塊

數據庫配置:
############    測試_內網      ############
#    host_name: '127.0.0.1'
#    port: 8080
#    service_name: 'service_name'
#    username: 'username'
#    password: 'password'
############    測試_外網      ############
    host_name: '127.0.0.1'
    port: 8080
    sid: 'sid'
    username: 'username'
    password: 'password'

 

import cx_Oracle
import os
import sys

sys.setrecursionlimit(10000)  # 例如這里設置為一萬
os.environ['NLS_LANG'] = 'SIMPLIFIED CHINESE_CHINA.UTF8'
os.environ['ORACLE_HOME'] = r'C:\Program Files\My Program Files\instantclient_12_2'
os.environ['TNS_ADMIN'] = r'C:\Program Files\My Program Files\instantclient_12_2'
# sys.path.insert(0, r'C:\Program Files\My Program Files\instantclient_12_2;')

def get_yamlMsg(sql_path):
    '''
    open方法打開直接讀出來
    :param sql_path:
    :return:
    '''
    with open(sql_path, 'r', encoding='utf-8') as f:
        sqlMsg = f.read()
    # 用load方法轉字典
    sql_dic = yaml.load(sqlMsg)
    return sql_dic

# 獲取數據庫配置文件路徑
db_conf_path = os.path.join(os.path.dirname(os.path.dirname(os.path.realpath(__file__))), 'conf', "application.yaml")
# 獲取數據庫配置信息
db_conf = get_yamlMsg(db_conf_path).get('數據庫配置')


''' 公共類'''
class MyOracle(object):
    def __init__(self):
        self.hostname = db_conf.get('host_name')
        self.port = db_conf.get('port')
        self.sid = db_conf.get('sid', None)
        self.service_name = db_conf.get('service_name', None)
        self.username = db_conf.get('username')
        self.password = db_conf.get('password')

    def get_connection(self):
        try:
            if self.service_name:
                tnsname = cx_Oracle.makedsn(self.hostname, self.port, service_name=self.service_name)
            else:
                tnsname = cx_Oracle.makedsn(self.hostname, self.port, self.sid)
            self.con = cx_Oracle.connect(self.username, self.password, tnsname)
        except Exception as e:
            print('連接數據庫出錯：', e)

    def get_cur(self):
        return self.con.cursor()

    def excute_one(self, sql):
        try:
            cur = self.get_cur()
            cur.execute(sql)
            # cur.prepare('select * from t_emp a where a.empid=:id')
            # cur.execute(None,{'id':id})
            line = cur.fetchone()
        except Exception as e:
            print('數據庫查詢出錯：', e)
            cur.close()
            self.con_close()
        cur.close()
        return line

    def excute_some(self, sql):
        try:
            cur = self.get_cur()
            cur.execute(sql)
            lines = cur.fetchall()
        except Exception as e:
            print('數據庫查詢出錯：', e)
            cur.close()
            self.con_close()
        cur.close()
        return lines

    def excute(self, sql):
        try:
            cur = self.get_cur()
            cur.execute(sql)
            effectRow = cur.rowcount
        except Exception as e:
            print('數據庫出錯：', e)
            cur.close()
            self.con_close()
        cur.close()
        return effectRow


    def con_close(self):
        self.con.close()

if __name__ == '__main__':
    db = MyOracle()
    db.get_connection()
    print(db.excute_some('''select * from dule'''))
    db.con_close()