python3连接Oracle11的工具类

刺骨的言语ヽ痛彻心扉 2022-10-23 12:44 164阅读 0赞

**目录**

一、环境准备

1、python3安装

2、Oracle11安装

二、安装驱动

1、pymogo驱动安装

2、pymogo驱动测试

三、py工具类

1、工具脚本

2、运行结果

--------------------

## 一、环境准备 ##

Oracle官网：[https://www.oracle.com/index.html][https_www.oracle.com_index.html]

CX\_Oracle：[https://oracle.github.io/python-cx\_Oracle/][https_oracle.github.io_python-cx_Oracle]

Oracle11：[https://docs.oracle.com/cd/E11882\_01/index.htm][https_docs.oracle.com_cd_E11882_01_index.htm]

Oracle SQL：[https://www.oracle.com/technetwork/cn/database/database-technologies/sql/documentation/index.html][https_www.oracle.com_technetwork_cn_database_database-technologies_sql_documentation_index.html]

Oracle+Python：[https://developer.oracle.com/cn/dsl/mastering-oracle-python.html][https_developer.oracle.com_cn_dsl_mastering-oracle-python.html]

Oracle教程：[https://www.w3cschool.cn/oraclejc/][https_www.w3cschool.cn_oraclejc]

Oracle语法：[http://www.quanxue.cn/qt\_xiaoya/commandindex.html][http_www.quanxue.cn_qt_xiaoya_commandindex.html]

### 1、python3安装 ###

windows离线安装python3.6.8环境：[https://blog.csdn.net/qq262593421/article/details/111309116][https_blog.csdn.net_qq262593421_article_details_111309116]

### 2、Oracle11安装 ###

Linux 离线安装 python 3.6.10：[https://blog.csdn.net/qq262593421/article/details/111309116][https_blog.csdn.net_qq262593421_article_details_111309116]

win10安装Oracle11：[https://blog.csdn.net/qq262593421/article/details/114324862][https_blog.csdn.net_qq262593421_article_details_114324862]

## 二、安装驱动 ##

### 1、pymogo驱动安装 ###

pip install  cx_Oracle

### 2、pymogo驱动测试 ###

python
    import cx_Oracle

## 三、py工具类 ##

### 1、工具脚本 ###

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()

### 2、运行结果 ###

> 11.2.0.1.0  
> FALSE  
> id  
> name  
> age

[https_www.oracle.com_index.html]: https://www.oracle.com/index.html
[https_oracle.github.io_python-cx_Oracle]: https://oracle.github.io/python-cx_Oracle/
[https_docs.oracle.com_cd_E11882_01_index.htm]: https://docs.oracle.com/cd/E11882_01/index.htm
[https_www.oracle.com_technetwork_cn_database_database-technologies_sql_documentation_index.html]: https://www.oracle.com/technetwork/cn/database/database-technologies/sql/documentation/index.html
[https_developer.oracle.com_cn_dsl_mastering-oracle-python.html]: https://developer.oracle.com/cn/dsl/mastering-oracle-python.html
[https_www.w3cschool.cn_oraclejc]: https://www.w3cschool.cn/oraclejc/
[http_www.quanxue.cn_qt_xiaoya_commandindex.html]: http://www.quanxue.cn/qt_xiaoya/commandindex.html
[https_blog.csdn.net_qq262593421_article_details_111309116]: https://blog.csdn.net/qq262593421/article/details/111309116
[https_blog.csdn.net_qq262593421_article_details_114324862]: https://blog.csdn.net/qq262593421/article/details/114324862