You are viewing pyhedgehog

Mon, Feb. 27th, 2012, 06:46 pm
Модуль dbms_output для cx_Oracle

Написал тут простенький модуль для использования вместе с 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.arraysize:
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 i<self.va.numElements: break
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.'


Wed, Feb. 29th, 2012 08:20 am (UTC)
pyvanet

Ты, оказывается, питонщик?
А слабо написать мне питоновую демку под вот такой OLE-control: http://jazz-soft.net/demo/vba/index.html ?

Mon, Mar. 19th, 2012 06:08 pm (UTC)
pyhedgehog

Сейчас посмотрю. Извини, но только сейчас заметил комментарий...

Fri, Mar. 23rd, 2012 03:51 am (UTC)
pyvanet

Спасибо!
После небольшого траха удалось заставить это работать...
Выложу на сайт в ближайшем будущем.
Как тебя написать в credentials?

Fri, Mar. 23rd, 2012 03:56 pm (UTC)
pyhedgehog

Как есть. По английски я пишусь Michael Dubner, если тебя именно это интересует.
А можешь и никак не писать - это ж просто перевод.

Sat, Mar. 24th, 2012 07:13 am (UTC)
pyvanet

Заапдейтил...