Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- 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()
Add Comment
Please, Sign In to add comment