A minimal sqlplus clone in IronPython

Hier folgt der Code für ein einfaches IronPython Programm, welches interaktiv die Eingabe von Sql -Statements und das Ausführen auf Oracle Datenbanken ermöglicht. Ich verwende den ODP, offenbar ist er bei Oracle 10g XE dabei, denn ich habe ihn nicht extra installiert. Außerdem beziehe ich mich auf hr das bekannte demo Schema.

Nach dem Start könnte eine Beispielsession wie folgt aussehen:

C:\D\FePy\sqldb>ipy uifs_oda_public.py
Data Source: xe
Username: hr
password: hr

ServerVersion: 10.2.0.1.0

Connection state: Open
hr@xe -->>
select table_name
from user_tables

> select table_name
from user_tables



TABLE_NAME REGIONS

TABLE_NAME LOCATIONS

TABLE_NAME DEPARTMENTS



und nun der Python Code:

------------------------- uifs_oda_public.py -----------------------------------------

'''
Bernd Kriszio: 2006-11-05
'''
import clr
import sys
clr.AddReference("System.Data")
clr.AddReference("Oracle.DataAccess")

import System.Data
import Oracle.DataAccess.Client

class ConnectionProperties :
def set(self, string):
self._dict = {}
for i in string.split(';'):
k, v = i.split('=')
self._dict[k.lower()] = v
def __getitem__(self, key):
return self._dict[key]

def n2onoff(val):
if val:
return 'ON'
else:
return 'OFF'

class Connection:
def __init__(self, ConnectString = ''):
self._echo = 0
self._feedback = 0
self._prompt = '>'
if not ConnectString:
ConnectString = self.queryConnectInfo()
self.Open(ConnectString)

def queryConnectInfo(self):
dataSource = raw_input('Data Source: ')
user = raw_input('Username: ')
password = raw_input('password: ')
return 'Data Source=%s;User ID=%s;Password=%s' % (dataSource, user, password)


def Set(self, OptionString):
''' some resamblance to SQL*PLUS
Tablemode on|off|auto
'''
s = OptionString.upper().strip()
if s == 'ECHO ON':
self._echo = 1
elif s == 'ECHO OFF':
self._echo = 0
elif s == 'FEEDBACK ON':
self._feedback = 1
elif s == 'FEEDBACK OFF':
self._feedback = 0

def Show(self, OptionString):
''' when there are more options i'll work it out, meanwhile SHOW ALL
'''
print 'ECHO is ' + n2onoff(self._echo)
print 'FEEDBACK is ' + n2onoff(self._echo)


def Open(self, ConnectString):
self._ConnectString = ConnectString
self._ConnectionProperties = ConnectionProperties()
self._ConnectionProperties.set(self._ConnectString)
self._Connection = Oracle.DataAccess.Client.OracleConnection(self._ConnectString)
self._Connection.Open()

print '\nServerVersion: ' + self._Connection.ServerVersion
print '\nConnection state: %s' % self._Connection.State.ToString()

def Close(self):
self._Connection.Close()

## def getConnection(self):
## return self._Connection

def setCommand(self, query):
self._Command = Oracle.DataAccess.Client.OracleCommand(query, self._Connection)

def ExecuteReader(self):
self._Reader = self._Command.ExecuteReader()
## bei SELECT immer -1
## if self._feedback == 1:
## print self._Reader.RecordsAffected

def ExecuteNonQuery(self):
self._count = self._Command.ExecuteNonQuery()
if self._feedback == 1:
print '%d rows affected' % self._count

def ExecSql(self, query):
if self._echo:
print '> ' + query
try:
## if 1:
if query.strip()[:6].upper() == 'SELECT':
self.setCommand(query)
self.ExecuteReader()
self.print_Reader()
elif query.strip()[:5].upper() == 'EXEC ':
statement = query[5:]
print 'statement %s' % statement
proc = statement.split(' ')[0]
print 'proc %s' % proc
elif query.strip()[:5].upper() == 'DESC ':
## print '--> desc'
statement = query[5:]
self.desc(statement.upper().strip())
else:
## print '--> else'
self.setCommand(query)
self.ExecuteNonQuery()
## if 0:
except StandardError, e:
if self._echo == 0:
print query
print "Fehler: ", e


def print_Reader (self):
rdr = self._Reader
if rdr:
anz = rdr.FieldCount
print
cnt = 0
while rdr.Read():
cnt += 1
for i in range(anz):
## print '%-30s %-30s%s' % (rdr.GetName(i), rdr.GetFieldType(i), rdr[i])
print '%-30s %s' % (rdr.GetName(i), rdr[i])
print
self._count = cnt
if self._feedback == 1:
print '%d rows affected' % self._count
rdr.Close()
rdr.Dispose()
self._Reader = None


def prompt1(self):
return '%s@%s -->>' % (self._ConnectionProperties['user id'], self._ConnectionProperties['data source'])

def ufi2(self):
''' multi line commands
I would like to use cancel ^C for cancel,
but KeyboardInterrupt works somewhat different in IronPython
'''
cmd = ''
print self.prompt1()
while 1:
a = raw_input()
try:
a_lc = a.lstrip().lower().split()[0]
except:
a_lc = ''
## work around. I would like to cancel input with ^C
if a_lc == 'cancel':
print 'input canceld'
cmd = ''
print self.prompt1()
elif a_lc == 'set':
self.Set(' '.join(a.lstrip().split()[1:]))
elif a_lc == 'show':
self.Show(' '.join(a.lstrip().split()[1:]))
elif a_lc == 'connect':
## print 'a:', a
p = ' '.join(a.lstrip().split()[1:])
## print 'p: ', p
try:
(up, dataSource) = p.split('@', 1)
except:
up = p
dataSource = self._ConnectionProperties['data source']

print up, dataSource

try:
user, password = up.split('/',1)
except:
user = up
password = raw_input('password: ')
connectString = 'Data Source=%s;User ID=%s;Password=%s' % (dataSource, user, password)
print '--> connect %s' % connectString
self.Close()
self.Open(connectString)
else:
cmd += a + '\n'
if cmd.lower() == 'quit\n':
break
if a == '' and cmd.strip():
## print 'calling: %s' % cmd
self.ExecSql(cmd)
else:
continue
cmd = ''
print self.prompt1()

## ----- Self test call -------------------------------------------------------------------------
if __name__=='__main__':

conn = Connection()
conn.Set('Echo on')
conn.Set('Feedback on')
conn.ufi2()
conn.Close()