python 访问 oralce 和mysql 的两个工具类

甩祸

oracle:  python test_python_cx_Oracle.py
renxp renxp 172.17.18.164:1521/orcl
 

#!/usr/bin/python
# Copyright Platform Computing Inc., an IBM company, 2012

import sys  
import string   
import os  
import time
import cx_Oracle as oraledb
# -----------------------------------------
# utility functions
# -----------------------------------------
class ISFDB:
"""
A class for handling the ISF database connection.
"""
def __init__(self, user=None, passwd=None, url=None):
"""
Init
"""
self.dbuser = user
self.dbpasswd = passwd
self.dburl  = url
self.__dbconn = None
if self.dbuser:
return
def connect(self, exit=True):
""" Connect to ISF database
"""
try:
self.__dbconn = oraledb.connect(self.dbuser, self.dbpasswd, self.dburl)
except Exception, e:
print "Failed to connect to ISF database %s" %self.dburl
print e
if exit:
sys.exit(1)
else:
return None
else:
#no exception occurred - obtain cursor
return self.__dbconn
def isconnected(self):
if self.__dbconn == None:
return False
else:
return True
def disconnect(self):
"""
Disconnect from the database
"""
if not self.isconnected():
return
self.__dbconn.close()
self.__dbconn = None
def checkConnection():
isf_dbuser  = sys.argv[1]
isf_dbpasswd = sys.argv[2]
isf_dburl = sys.argv[3]
# Connect to the database
database = ISFDB(isf_dbuser, isf_dbpasswd, isf_dburl)
database.connect()
database.disconnect()
return
if __name__ == "__main__":
checkConnection()
sys.exit(0)
 
　　
MySQL：python
test_python_MySQL.py 172.17.27.242 3306 root
111111 test
 

#!/usr/bin/python
# Copyright Platform Computing Inc., an IBM company, 2012

import sys  
import string   
import os  
import time
import MySQLdb as mysqldb
# -----------------------------------------
# utility functions
# -----------------------------------------
class ISFDB:
"""
A class for handling the ISF database connection.
"""
def __init__(self, host=None,port=None,user=None, passwd=None, dbname=None):
"""
Init
"""
self.dbhost = host
self.dbport = port
self.dbuser = user
self.dbpasswd = passwd
self.dbname = dbname
self.__dbconn = None
if self.dbuser:
return
def connect(self, exit=True):
""" Connect to ISF database
"""
try:
self.__dbconn = mysqldb.connect(host='%s' %self.dbhost, port=string.atoi(self.dbport),user='%s' %self.dbuser,\
passwd='%s' %self.dbpasswd, db='%s' %self.dbname)
except Exception, e:
print "Failed to connect to ISF database %s" %self.dbhost
print e
if exit:
sys.exit(1)
else:
return None
else:
#no exception occurred - obtain cursor
return self.__dbconn
def isconnected(self):
if self.__dbconn == None:
return False
else:
return True
def disconnect(self):
"""
Disconnect from the database
"""
if not self.isconnected():
return
self.__dbconn.close()
self.__dbconn = None
def checkConnection():
isf_dbhost = sys.argv[1]
isf_dbport = sys.argv[2]
isf_dbuser  = sys.argv[3]
isf_dbpasswd = sys.argv[4]
isf_dbname = sys.argv[5]
# Connect to the database
database = ISFDB(isf_dbhost, isf_dbport, isf_dbuser,isf_dbpasswd,isf_dbname)
database.connect()
database.disconnect()
return
if __name__ == "__main__":
checkConnection()
sys.exit(0)