python3连接Oracle11的工具类

pip install  cx_Oracle

python
import cx_Oracle

import cx_Oracle

class OracleUtil:
    """
    Oracle工具类
    """
    def __init__(self, url=None, user=None, passwd=None, ip="127.0.0.1", port=1521, sid="orcl", encoding="UTF-8"):
        """构造函数"""
        if (url is not None):
            self.__url = url
        else:
            self.__url = user + "/" + passwd + "@" + ip + ":" + str(port) + "/" + sid
        self.__conn = cx_Oracle.connect(self.__url, encoding=encoding)
        self.__cursor = self.__conn.cursor()

    def __del__(self):
        """析构函数"""
        self.__cursor.close()
        self.__conn.close()

    def get_conn(self):
        """获取连接"""
        return self.__conn

    def get_cursor(self):
        """获取游标"""
        return self.__cursor

    def get_version(self):
        """获取Oracle版本"""
        return self.__conn.version

    def get_version_info(self):
        """获取Oracle版本"""
        self.__cursor.execute("SELECT * FROM v$version")
        result = []
        for row in self.__cursor.fetchall():
            result.append(row[0])
        return result

    def create_user(self, user, passwd):
        """创建新用户"""
        self.__cursor.execute("CREATE USER " + str(user) + " IDENTIFIED BY " + str(passwd))

    def list_user_simple_info(self):
        """查询所有用户信息"""
        self.__cursor.execute("SELECT * FROM all_users")
        return self.__cursor.fetchall()

    def list_user_info(self):
        """查询所有用户信息"""
        self.__cursor.execute("SELECT * FROM dba_users")
        return self.__cursor.fetchall()

    def is_dba(self):
        """判断用户是否为dba"""
        self.__cursor.execute("SELECT SYS_CONTEXT('USERENV','ISDBA') FROM dual")
        return self.__cursor.fetchall()[0][0]

    def list_tablespaces(self):
        """查询表空间（Oracle没有数据库）"""
        self.__cursor.execute("SELECT NAME FROM v$tablespace")
        tablespaces = []
        for tablespace in self.__cursor.fetchall():
            # yield table[0]
            tablespaces.append(tablespace[0])
        return tablespaces

    def list_tables(self):
        """查询所有表"""
        self.__cursor.execute("SELECT TABLE_NAME FROM all_tables")
        tables = []
        for table in self.__cursor.fetchall():
            # yield table[0]
            tables.append(table[0])
        return tables

    def list_views_info(self):
        """查询所有视图信息"""
        self.__cursor.execute("SELECT * FROM all_views")
        views = []
        for view in self.__cursor.fetchall():
            views.append(view)
        return views

    def list_views(self):
        """查询所有视图"""
        self.__cursor.execute("SELECT VIEW_NAME FROM all_views")
        views = []
        for view in self.__cursor.fetchall():
            views.append(view[0])
        return views

    def list_user_views_info(self):
        """查询当前用户所有视图信息"""
        self.__cursor.execute("SELECT * FROM user_views")
        views = []
        for view in self.__cursor.fetchall():
            views.append(view)
        return views

    def list_user_views(self):
        """查询当前用户所有视图名称"""
        self.__cursor.execute("SELECT VIEW_NAME FROM user_views")
        views = []
        for view in self.__cursor.fetchall():
            views.append(view[0])
        return views

    def list_roles(self):
        """查询所有角色"""
        self.__cursor.execute("SELECT ROLE FROM dba_roles")
        roles = []
        for role in self.__cursor.fetchall():
            roles.append(role[0])
        return roles

    def get_user_table(self):
        """查询当前数据库的所有表名"""
        self.__cursor.execute("SELECT * FROM user_tables")
        tables = []
        for table in self.__cursor.fetchall():
            # yield table[0]
            tables.append(table[0])
        return tables

    def get_database_info(self):
        """获取当前数据库信息"""
        self.__cursor.execute("SELECT * FROM v$database")
        return self.__cursor.fetchall()

    def get_instance_info(self):
        """获取数据库实例"""
        self.__cursor.execute("SELECT * FROM v$instance")
        return self.__cursor.fetchall()

    def get_table_columns_info(self, table):
        """查询表字段信息"""
        self.__cursor.execute("SELECT * FROM user_tab_columns WHERE table_name='"+ table +"'")
        return self.__cursor.fetchall()

    def get_table_columns(self, table):
        """查询表字段名"""
        self.__cursor.execute("SELECT COLUMN_NAME FROM user_tab_columns WHERE table_name='"+ table +"'")
        columns = []
        for column in self.__cursor.fetchall():
            columns.append(column[0])
        return columns

    def execute(self, sql):
        """执行SQL"""
        self.__cursor.execute(sql)
        return self.__cursor.fetchall()

    def list_table_metadata(self):
        """查询所有表的元数据信息"""
        sql = "SELECT * FROM information_schema.TABLES WHERE TABLE_TYPE !='SYSTEM VIEW' AND TABLE_SCHEMA NOT IN ('sys','mysql','information_schema','performance_schema')"
        self.__cursor.execute(sql)
        return self.__cursor.fetchall()

    def get_table_fields(self, db, table):
        """获取表字段信息"""
        sql = 'SELECT COLUMN_NAME FROM information_schema.COLUMNS WHERE table_schema="' + db + '" AND table_name="' + table + '"'
        cursor = self.conn.cursor()
        self.__cursor.execute(sql)
        fields = []
        for field in cursor.fetchall():
            fields.append(field[0])
        return fields

    def table_metadata(self, db, table):
        """查询表字段的元数据信息"""
        db = "'" + db + "'"
        table = "'" + table + "'"
        sql = """
        SELECT 
            column_name,column_type,ordinal_position,column_comment,column_default 
        FROM 
            information_schema.COLUMNS 
        WHERE 
            table_schema = %s AND table_name = %s;
        """ % (db, table)
        self.__cursor.execute(sql)
        return self.__cursor.fetchall()


if __name__ == "__main__":
    url = 'xl/123456@192.168.81.166:1521/orcl'
    oracleUtil = OracleUtil(url=url)
    print(oracleUtil)
    print(oracleUtil.get_version())
    # for info in oracleUtil.get_version_info():
    #     print(info)
    print(oracleUtil.is_dba())
    # ## 查询所有用户信息
    # for user_info in oracleUtil.list_user_simple_info():
    #     print(user_info)
    ## 不带参数的函数
    for table in oracleUtil.list_roles():
        print(table)
    ## 带参数的函数
    for table in oracleUtil.get_table_columns('t_user'):
        print(table)
    ## 创建用户
    # print(oracleUtil.create_user("xl02", "123456"))
    # oracleUtil.list_databases()