这个项目的地址 传送门
第一个python3项目,对python 还是学习中,请大佬轻喷,欢迎指点
import jaydebeapifrom .database import Databasedirver = 'oracle.jdbc.OracleDriver'
jarFile = '/Users/warrior/Code/python/builder/jar/ojdbc6-11.2.0.1.0.jar'class DbOracle(Database):def get_jdbc_connection(self):import jpypeif jpype.isJVMStarted() and not jpype.isThreadAttachedToJVM():jpype.attachThreadToJVM()jpype.java.lang.Thread.currentThread().setContextClassLoader(jpype.java.lang.ClassLoader.getSystemClassLoader())connection = jaydebeapi.connect(dirver, self.url, {'user': self.user, 'password': self.password, 'tmode': 'TERA', 'charset': 'UTF8'}, jarFile)return connectiondef connect(self, url, user, password):self.url = urlself.user = userself.password = passworddef userTables(self):data = self.execute("""SELECT A.TABLE_NAME, B.COMMENTS FROM USER_TABLES A LEFT JOIN USER_TAB_COMMENTS B ON A.TABLE_NAME=B.TABLE_NAME""")result = []for table in data:result.append({'name': table[0], 'comment': table[1]})return resultdef columns(self, table_name):data = self.execute("""SELECT A.COLUMN_NAME, A.DATA_TYPE, A.DATA_LENGTH, A.DATA_PRECISION, A.DATA_SCALE, A.NULLABLE, B.COMMENTS FROM USER_TAB_COLUMNS A LEFT JOIN USER_COL_COMMENTS B ON A.TABLE_NAME=B.TABLE_NAME AND A.COLUMN_NAME=B.COLUMN_NAME WHERE A.TABLE_NAME='%s' ORDER BY COLUMN_ID ASC""" % table_name)result = []primary_key = self.table_primary(table_name)for column in data:result.append({'name': column[0], 'type': column[1], 'primary': column[0] in primary_key,'length': column[2], 'precision': column[3], 'scale': column[4], 'nullable':column[5], 'comment': column[6]})return resultdef table_primary(self, table_name):sql = """SELECT cols.column_name FROM all_constraints cons, all_cons_columns colsWHERE cols.table_name = '%s' AND cons.constraint_type = 'P'AND cons.constraint_name = cols.constraint_name AND cons.owner = cols.ownerORDER BY cols.table_name""" % table_namedata = self.execute(sql)result = []for d in data:result.append(d[0])return resultdef execute(self, sql):conn = self.get_jdbc_connection()curs = conn.cursor()curs.execute(sql)result = curs.fetchall()curs.close()conn.close()return resultdef close(self):if (self.conn != None):self.conn.close()