Модуль dbms_output для cx_Oracle

[pyhedgehog]

Написал тут простенький модуль для использования вместе с cx_Oracle для получения вывода через модуль dbms_output.


Пример:

import cx_Oracle
import dbms_output
db=cx_Oracle.connect(connect_string)
output=dbms_output.dbms_output(db)
c=db.cursor()
#вывод попадёт в stdout
c.callproc("dbms_output.put_line",("stdout test",))
output.disable()
#вывод будет проигнорирован
c.callproc("dbms_output.put_line",("ignored test",))
db.close()
Чуть более сложный пример:

import cx_Oracle
import logging
import dbms_output

log = logging.getLogger("dbms_output")

class log_dbms_output(dbms_output):
def output(self, s):
log.info("%s", s)

db=cx_Oracle.connect(connect_string)
output=log_dbms_output(db)
c=db.cursor()
#вывод попадёт в log (в зависимости от настроек модуля logging)
c.callproc("dbms_output.put_line",("log test",))
output.disable()
#вывод будет проигнорирован
c.callproc("dbms_output.put_line",("ignored test",))
db.close()


Ну и сам модуль dbms_output.py:

import threading

FNCODE_STMTEXECUTE = 21
UCBTYPE_EXIT = 2

class dbms_output:
def __init__(self, db, arraysize=50):
self.db = db
self.lck = threading.Lock()
self.enabled = False
self.arraysize = arraysize or 1
self.c = None
self.va = None
self.vs = None
self.vi = None
self.enable()
def catch(self, *a):
if not self.enabled: return
l = self.lck
if not l.acquire(0): return
try:
while True:
if self.arraysize>1:
if self.va is None or self.va.numElements self.vs = None
self.va = self.c.arrayvar(str,self.arraysize)
self.vi.setvalue(0,self.va.numElements)
a,i = self.c.callproc("dbms_output.get_lines",(self.va,self.vi))
for s in a[:i]:
self.output(s or '')
if ibreak
else:
if self.vs is None:
self.va = None
self.vs = self.c.var(str)
s,i = self.c.callproc("dbms_output.get_line",(self.vs,self.vi))
if i: break
self.output(s or '')
finally:
l.release()
def register(self):
self.db.register(FNCODE_STMTEXECUTE,UCBTYPE_EXIT,self.catch)
def unregister(self):
self.db.unregister(FNCODE_STMTEXECUTE,UCBTYPE_EXIT)
def enable(self):
if self.enabled: return
self.c = self.db.cursor()
self.c.callproc("dbms_output.enable")
if self.arraysize>1:
self.va = self.c.arrayvar(str,self.arraysize)
else:
self.vs = self.c.var(str)
self.vi = self.c.var(int)
self.register()
self.enabled = True
self.catch()
def disable(self):
if not self.enabled: return
self.unregister()
self.c.callproc("dbms_output.disable",())
self.c.close()
self.c = None
self.va = None
self.vs = None
self.vi = None
self.enabled = False
def output(self, s):
print s

if __name__=='__main__':
print 'Not a script.'