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()