函数方式封装

import cx_Oracle
import pprint,traceback


__version__=1.1

_ou_conn=None
_ou_rowcount=0
def connect(user,password,dsn,enconding='utf-8'):
    '''
    连接数据库
    user：用户名,
    password：密码,
    dsn：ip:port/sid,
    enconding：编码
    '''
    global _ou_conn
    _ou_conn=cx_Oracle.connect(user=user,password=password,dsn=dsn,encoding=enconding)
    # print('连接成功！')

def close():
    '''
    关闭连接
    '''
    _ou_conn.close()

def queryall(sql,params=None):
    '''
    查询，返回查询的所有结果，类型：list
    '''
    global _ou_rowcount
    rows=None
    try:
        cursor=_ou_conn.cursor()
        # 调用prepare()方法，发送sql，并对sql进行预编译
        cursor.prepare(sql)
        # 调用execute()，不要传递sql语句，否则预编译sql无效，执行的还是execute()方法的sql
        if params==None:
            cursor.execute(None)
        else:
            cursor.execute(None,params)
        rows=cursor.fetchall()
        _ou_rowcount=cursor.rowcount
    # 不要处理异常
    # except Exception as e:
    #     print(e)
    finally:
        cursor.close()
    return rows


def queryone(sql,params=None):
    '''
    查询，返回一条结果，类型：tuple
    '''
    global _ou_rowcount
    row=None
    try:
        cursor=_ou_conn.cursor()
        # 调用prepare()方法，发送sql，并对sql进行预编译
        cursor.prepare(sql)
        # 调用execute()，不要传递sql语句，否则预编译sql无效，执行的还是execute()方法的sql
        if params==None:
            cursor.execute(None)
        else:
            cursor.execute(None,params)
        row=cursor.fetchone()
        _ou_rowcount=cursor.rowcount
    # 不要处理异常
    # except Exception as e:
    #     print(e)
    finally:
        cursor.close()
    return row


def update(sql,params=None):
    '''
    执行insert、update、delete语句
    sql：insert、update、delete语句
    params：sql语句的参数
    '''
    global _ou_rowcount
    cursor=_ou_conn.cursor()
    try:
        # 调用prepare()方法，发送sql，并对sql进行预编译
        cursor.prepare(sql)
        if params:
            cursor.execute(None,params)
        else:
            cursor.execute(None)

        _ou_rowcount=cursor.rowcount

    finally:
        cursor.close()
    return _ou_rowcount

def commit():
    '''
    事务提交
    '''
    _ou_conn.commit()

def rollback():
    '''
    事务回滚
    '''
    _ou_conn.rollback()

测试代码：

if __name__ == "__main__":
    connect('scott','tiger','127.0.0.1:1521/oradb','utf-8')
    try:

        # result=queryall('select * from emp where ename=:ename and job=:job',{'ename':'ALLEN','job':'SALESMAN'} )
        # print('查询结果数：',_ou_rowcount)
        # pprint.pprint(result)
        # print('-'*50)
        # result=queryall('select * from dept where 1=2')
        # print('查询结果数：',_ou_rowcount)
        # pprint.pprint(result)

        # print('-'*50)

        # # result=queryall('select * from dept where deptno=:1',[10])
        # result=queryone('select * from dept where deptno=:1',[111110])
        # print('查询结果数：',_ou_rowcount)
        # pprint.pprint(result)

        params={'id':'1','snumber':'201908190001','total':'90'}
        num=update('insert into orders (id,snumber,total) values (:id,:snumber,:total)',params)
        print('影响的行数：',num)

        params={'id':'1','name':'篮球','price':'50'}
        update('insert into order_goods (id,name,price) values (:id,:name,:price)',params)
        print('影响的行数：',num)

        # params={'id':'2','name':'足球','price':'40'}
        update("insert into order_goods (id,name,price) values ('2','足球','40')")
        print('影响的行数：',num)

        commit()
    except BaseException as e:
        rollback()
        # 打印异常详细信息
        traceback.print_exc()
        # 将异常的详细信息，作为字符串返回
        # print(traceback.format_exc())
        print('报错')
    finally:
        close()

原文出处：/show_1EF3uYtMpczO.html