Instantly share code, notes, and snippets.

Embed
What would you like to do?
class OracleCursor:
"""游标类"""
def __init__(self, cursor, logger, dict_result):
self.cursor = cursor
self.logger = logger
self.dict_result = dict_result
def _dict_result(self, cursor):
cols = [d[0].lower() for d in cursor.description]
def row_factory(*args):
return dict(zip(cols, args))
return row_factory
def execute(self, sql, params=None):
if params:
self.cursor.execute(sql, params)
else:
self.cursor.execute(sql)
def query(self, sql, params=None, with_description=False):
if params:
self.cursor.execute(sql, params)
else:
self.cursor.execute(sql)
if self.dict_result:
self.cursor.rowfactory = self._dict_result(self.cursor)
rows = self.cursor.fetchall()
if with_description:
res = rows, self.cursor.description
else:
res = rows
return res
class OracleInstance:
"""实例类"""
def __init__(self, username, password, host, port, sid,
charset='utf8',dict_result=False,logger=None):
self.username = username
self.password = password
self.host = host
self.port = port
self.sid = sid
self.charset = charset
self.dict_result = dict_result
if logger is None:
logger=logging.getLogger(__name__)
self.logger = logger
def __enter__(self):
dsn_tns = cx_Oracle.makedsn(self.host, self.port, self.sid)
self.con = cx_Oracle.connect(self.username, self.password, dsn_tns)
self.cursor = self.con.cursor()
return OracleCursor(self.cursor, self.logger, self.dict_result)
def __exit__(self, exc_type, exc_val, exc_tb):
self.cursor.execute("commit")
self.cursor.close()
self.con.close()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment