Sunday, March 18, 2012

Robot Framework - Database Libraries

Robot framework already has two database libraries and both are implemented in different lanugauges.
Database Library - Java
Database Library - Python

I have developed two simple database libraries. One is for generic ODBC interface and another one is especially for Oracle. The second one can be used as thin client.
Database Library - ODBC interface

#
# Author: Palani Selvam
#
# NOTE: It is supported all type of Databases, which are having support to ODBC
# For ex, DB2, ORACLE, MS SQL SERVER, Sybase.
#
import pyodbc as DB
from robot.errors import DataError
from robot import utils

__version__ = '0.1' # Developed by Palani - 20101022

class DBLibrary:

ROBOT_LIBRARY_SCOPE = 'GLOBAL'
ROBOT_LIBRARY_VERSION = __version__

def __init__(self):
"""DBLibrary Library can be imported with optional arguments.

"""
self._connection = None
self._default_log_level = 'INFO'
self._bexception = False
#self._prompt = prompt

def make_connection(self, sDsn, sUid=None, sPwd=None):
""" make_connection method is used to create the connection object.
sDsn - DataSource Name
sUid - User Id to access the database
sPwd - Password to access the database
"""
sConn = None
try:
sConn = DB.connect(dsn=sDsn, uid=sUid, pwd=sPwd )

except Exception:
self._bexception = True
print 'Connection for database has failed', Exception
raise
finally:
self._connection = sConn

def execute_sql (self,sSql=None,iRows=None):
""" execute_sql method is used to execute the sql query to the given connection object.
sSql - Query to access the database
iRows - No of rows to be returned.
"""
sOut = None
sCursor = None
sConn = self._connection

if (sConn == None) :
return sOut
else:
try:
sCursor = sConn.cursor()
sCursor.execute(sSql )
# curs.execute("SELECT * FROM EmpTable WHERE SCOPE=3 ")
if (iRows <> None) :
iRows = int (iRows)
sOut = sCursor.fetchmany (iRows)
else :
sOut = sCursor.fetchall ()

except Exception:
self._bexception = True
print 'Query execution is failed. ', Exception
raise
finally:
if (sCursor <> None):
sCursor.close ()
return sOut

def execute_for_one_row (self,sSql=None):
""" execute_for_one_row method is used to execute the sql query and returns only first row.
sSql - Query to access the database
"""
sOut = None
sCursor = None
sConn = self._connection

if (sConn == None) :
return sOut
else:
try:
sCursor = sConn.cursor()
sCursor.execute(sSql )
rows = sCursor.fetchone ()
for sOut in rows:
break

except Exception:
self._bexception = True
print 'Query execution is failed. ', Exception
raise
finally:
if (sCursor <> None):
sCursor.close ()
return sOut

def disconnect_from_database (self,sConn=None):
""" disconnect_from_database method is used to disconnect from the given connection object.
sConn - Connected Object, which is got through make_connection method.
"""
if (sConn <> None):
sConn.close ()
else:
self._connection.close ()

Database Library - Oracle
#
# Author: Palani Selvam
#
#
# NOTE: It is supported only ORACLE Databases
# DSN is not required.
#
import cx_Oracle as DB
from robot.errors import DataError
from robot import utils

__version__ = '0.1' # Developed by Palani

class ORADBLibrary:

ROBOT_LIBRARY_SCOPE = 'GLOBAL'
ROBOT_LIBRARY_VERSION = __version__

def __init__(self):
"""DBLibrary Library can be imported with optional arguments.

"""
self._connection = None
self._default_log_level = 'INFO'
self._bexception = False
#self._prompt = prompt

def ora_make_connection(self, sConnString):
""" make_connection method is used to create the connection object.
sConnString - Connection String

Example:
'sqluser/welcome@127.0.0.1/orcl'
"""
sConn = None
try:
sConn = DB.connect(sConnString )

except Exception:
self._bexception = True
print 'Connection for database has failed', Exception, sConnString
raise
finally:
self._connection = sConn

def ora_execute_sql (self,sSql=None,iRows=None):
""" execute_sql method is used to execute the sql query to the given connection object.
sSql - Query to access the database
iRows - No of rows to be returned.
"""
sOut = None
sCursor = None
sConn = self._connection

if (sConn == None) :
return sOut
else:
try:
sCursor = sConn.cursor()
sCursor.execute(sSql )
# curs.execute("SELECT * FROM EMP_Table WHERE SCOPE=3 ")
if (iRows <> None) :
iRows = int (iRows)
sOut = sCursor.fetchmany (iRows)
else :
sOut = sCursor.fetchall ()

except Exception:
self._bexception = True
print 'Query execution is failed. ', Exception
raise
finally:
if (sCursor <> None):
sCursor.close ()
return sOut

def ora_execute_for_one_row (self,sSql=None):
""" execute_for_one_row method is used to execute the sql query and returns only first row.
sSql - Query to access the database
"""
sOut = None
sCursor = None
sConn = self._connection

if (sConn == None) :
return sOut
else:
try:
sCursor = sConn.cursor()
sCursor.execute(sSql )
rows = sCursor.fetchone ()
for sOut in rows:
break

except Exception:
self._bexception = True
print 'Query execution is failed. ', Exception
raise
finally:
if (sCursor <> None):
sCursor.close ()
return sOut

def ora_disconnect_from_database (self,sConn=None):
""" disconnect_from_database method is used to disconnect from the given connection object.
sConn - Connected Object, which is got through make_connection method.
"""
if (sConn <> None):
sConn.close ()
else:
self._connection.close ()
Like above, we can extend Robot Framework easily...

1 comment:

Anonymous said...

This is just what I am looking for.

Can you provide information on how to add this library into RobotFramework?

Thnx