自动化测试常用脚本-连接Oracle数据库

"""
用于连接Oracle数据库,使用前需确认系统已安装cx_Oracle库和oracle客户端
cx_Oracle就可以在PyPI中下载,https://pypi.python.org/pypi,根据Python版本和系统位数进行下载
Oracle客户端,百度找instantclient,根据系统确定好是下载32位的还是64位
"""
# -*- coding:utf-8 -*-
import cx_Oracle as Oracle
from DBUtils.PooledDB import PooledDB


class OrclPool(object):
    """
    1) 这里封装了一些有关oracle连接池的功能;
    2) sid和service_name,程序会自动判断哪个有值,
        若两个都有值,则默认使用sid;
        若只想用其中一个,则只需要把另一个设置为空即可。如,service_name = ‘‘
    3) 关于config的设置,只有 port 的值的类型是 int,其他均为str:
    """

    def __init__(self, config):
        self.conn = OrclPool.__get_conn(config)  # 获得连接池,Oracle连接信息
        self.cur = self.conn.cursor()  # 获取操作游标

    @staticmethod
    def __get_conn(conf):
        """
        :param config:        dict    连接Oracle的信息
        ————————————————————————
        maxconnections=6,   # 最大连接数,0或None表示不限制连接数
        mincached=2,        # 初始化时,连接池中至少创建的空闲连接。0表示不创建
        maxcached=5,        # 连接池中最多允许的空闲连接数,很久没有用户访问,连接池释放了一个,由6个变为5个,
                            # 又过了很久,不再释放,因为该项设置的数量为5
        maxshared=0,        # 在多个线程中,最多共享的连接数,Python中无用,会最终设置为0
        blocking=True,      # 没有闲置连接的时候是否等待, True,等待,阻塞住;False,不等待,抛出异常。
        maxusage=None,      # 一个连接最多被使用的次数,None表示无限制
        setession=[],       # 会话之前所执行的命令, 如["set charset ...", "set datestyle ..."]
        ping=0,             # 0  永远不ping
                            # 1,默认值,用到连接时先ping一下服务器
                            # 2, 当cursor被创建时ping
                            # 4, 当SQL语句被执行时ping
                            # 7, 总是先ping

        """
        host, port, sid, service_name = conf.get(‘host‘), conf.get(‘port‘), conf.get(‘sid‘), conf.get(‘service_name‘)

        dsn = None
        if sid:
            dsn = Oracle.makedsn(host, port, sid=sid)
        elif service_name:
            dsn = Oracle.makedsn(host, port, service_name=conf.get(‘service_name‘))

        __pool = PooledDB(Oracle, user=conf[‘user‘], password=conf[‘passwd‘], dsn=dsn, mincached=1, maxcached=5)
        return __pool.connection()

    def execute_sql(self, sql, args=None):
        """
        执行sql语句
        :param sql:     str     sql语句
        :param args:    list    sql语句参数列表
        """
        if args:
            self.cur.execute(sql, args)
        else:
            self.cur.execute(sql)

    def fetch_all(self, sql, args=None):
        """
        获取全部结果
        :param sql:     str     sql语句
        :param args:    list    sql语句参数
        :return:        tuple   fetch结果
        """
        self.execute_sql(sql, args)  # 执行 SQL 语句
        return self.cur.fetchall()   # 返回执行的结果

   def fetch_one(self, sql, args=None):
        """
        获取全部结果
        :param sql:     str     sql语句
        :param args:    list    sql语句参数
        :return:        tuple   fetch结果
        """
        conn, cursor = self.__execute(sql, args)
        result = cursor.fetchone()
        self.__reset_conn(conn, cursor)
 
        return result

    def commit_sql(self, sql):
        self.cur.execute(sql)
        self.conn.commit()  # 用于SQL语句提交事务

    def __del__(self):
        """
        在实例资源被回收时,关闭该连接池与数据库
        """
        self.cur.close()
        self.conn.close()


if __name__ == "__main__":
    orcl_cfg = {
        ‘user‘: ‘username‘,  # 数据库用户名字
        ‘passwd‘: ‘password‘,  # 数据库密码
        ‘host‘: ‘0.0.0.0‘,  # 数据库链接地址
        ‘port‘: 3000,  # 端口
        ‘sid‘: ‘‘,
        ‘service_name‘: ‘TESTDB‘,  # 数据库名字
    }
    orcl = OrclPool(orcl_cfg)



部分内容参考:https://blog.csdn.net/maixiaochai/article/details/82986517

自动化测试常用脚本-连接Oracle数据库

上一篇:未找到具有固定名称“System.Data.SQLite”的 ADO.NET 提供程序的实体框架提供程序


下一篇:Oracle 多行数据合并成一行