• Python 连接 Oracle 详解

    文章目录

    1 概述

    1.1 安装第三方库 cx_Oracle

    1.2 测试连接

    import cx_Oracle

# 1.基本信息
user = 'scott'  # 用户名
password = '123456'  # 密码
host = '127.0.0.1:1521'  # 主机:端口
instance = 'orcl'  # 服务名

# 2.格式：user/password@host/service_name
connect_str = f'{user}/{password}@{host}/{instance}'

# 3.测试
try:
    connect_info = cx_Oracle.connect(connect_str)
    print(f'[提示]连接成功，信息：{connect_info}')
except Exception as e:
    print(f'[提示]连接失败，报错：{e}')

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17

    执行结果：

    <cx_Oracle.Connection to scott@127.0.0.1:1521/orcl>

    • 1

    2 常用操作

    2.1 查询

    2.1.1 查询所有记录

    import cx_Oracle


class Oracle(object):
    def __init__(self, user_name, password, host, instance):
        self._conn = cx_Oracle.connect(f'{user_name}/{password}@{host}/{instance}')
        self.cursor = self._conn.cursor()
        print('[提示]: 连接成功')

    def query_all(self, sql):
        """查询所有记录"""
        self.cursor.execute(sql)
        return self.cursor.fetchall()


if __name__ == '__main__':
    test = Oracle('scott', '123456', '127.0.0.1:1521', 'orcl')
    
    # 测试：查询所有记录
    test_sql = 'select * from scott.dept'
    result = test.query_all(test_sql)
    print(result)

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22

    执行结果：

    [提示]: 连接成功
[(10, 'ACCOUNTING', 'NEW YORK'), 
 (20, 'RESEARCH', 'DALLAS'), 
 (30, 'SALES', 'CHICAGO'), 
 (40, 'OPERATIONS', 'BOSTON')]

    • 1
    • 2
    • 3
    • 4
    • 5

    2.1.2 查询单条记录

    import cx_Oracle


class Oracle(object):
    def __init__(self, user_name, password, host, instance):
        self._conn = cx_Oracle.connect(f'{user_name}/{password}@{host}/{instance}')
        self.cursor = self._conn.cursor()
        print('[提示]: 连接成功')

    def query_one(self, sql):
        """查询单条记录"""
        self.cursor.execute(sql)
        return self.cursor.fetchone()


if __name__ == '__main__':
    test = Oracle('scott', '123456', '127.0.0.1:1521', 'orcl')
    test_sql = 'select * from scott.dept'
    
    # 测试：查询单条记录
    result = test.query_one(test_sql)
    print(result)

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22

    执行结果：

    [提示]: 连接成功
(10, 'ACCOUNTING', 'NEW YORK')

    • 1
    • 2

    2.1.3 根据条件查询记录

    import cx_Oracle


class Oracle(object):
    def __init__(self, user_name, password, host, instance):
        self._conn = cx_Oracle.connect(f'{user_name}/{password}@{host}/{instance}')
        self.cursor = self._conn.cursor()
        print('[提示]: 连接成功')

    def query_by(self, sql, params):
        """根据条件查询记录"""
        self.cursor.execute(sql, params)
        return self.cursor.fetchall()


if __name__ == '__main__':
    test = Oracle('scott', '123456', '127.0.0.1:1521', 'orcl')

    # 测试：根据条件查询记录
    result = 'select * from scott.dept where deptno = :deptno and dname = :dname'
    result = test.query_by(result, {'deptno': 10, 'dname': 'ACCOUNTING'})
    print(result)

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22

    执行结果：

    [提示]: 连接成功
[(10, 'ACCOUNTING', 'NEW YORK')]

    • 1
    • 2

    2.2 新增

    import cx_Oracle


class Oracle(object):
    def __init__(self, user_name, password, host, instance):
        self._conn = cx_Oracle.connect(f'{user_name}/{password}@{host}/{instance}')
        self.cursor = self._conn.cursor()
        print('[提示]: 连接成功')

    def insert(self, sql, params):
        """插入数据"""
        self.cursor.execute(sql, params)
        self._conn.commit()
        print('[提示]：执行成功！')


if __name__ == '__main__':
    test = Oracle('scott', '123456', '127.0.0.1:1521', 'orcl')

    # 测试
    test_seq = "insert into scott.dept_bak(deptno, dname, loc) values(:deptno, :dname, :loc)"
    test.insert(test_seq, {'deptno': 50, 'dname': 'a', 'loc': 'b'})

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22

    2.3 修改

    import cx_Oracle


class Oracle(object):
    def __init__(self, user_name, password, host, instance):
        self._conn = cx_Oracle.connect(f'{user_name}/{password}@{host}/{instance}')
        self.cursor = self._conn.cursor()
        print('[提示]: 连接成功')

    def update(self, sql, params):
        """修改数据"""
        self.cursor.execute(sql, params)
        self._conn.commit()
        print('[提示]: 修改完成！')


if __name__ == '__main__':
    test = Oracle('scott', '123456', '127.0.0.1:1521', 'orcl')

    # 测试
    test_seq = "update scott.dept_bak t set t.dname = :dname, t.loc = :loc where t.deptno = :deptno"
    test.update(test_seq, {'deptno': 50, 'dname': 'aaa', 'loc': 'bbbb'})

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22

    2.4 删除

    2.4.1 清空表

    import cx_Oracle


class Oracle(object):
    def __init__(self, user_name, password, host, instance):
        self._conn = cx_Oracle.connect(f'{user_name}/{password}@{host}/{instance}')
        self.cursor = self._conn.cursor()
        print('[提示]: 连接成功')

    def truncate(self, table_name):
        """清空表数据"""
        sql = 'truncate table ' + table_name
        self.cursor.execute(sql)
        self._conn.commit()
        print(f'[提示]: 清空表 {table_name} 成功')


if __name__ == '__main__':
    test = Oracle('scott', '123456', '127.0.0.1:1521', 'orcl')

    # 测试
    test.truncate('scott.dept_bak')

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22

    2.4.2 删除表

    import cx_Oracle


class Oracle(object):
    def __init__(self, user_name, password, host, instance):
        self._conn = cx_Oracle.connect(f'{user_name}/{password}@{host}/{instance}')
        self.cursor = self._conn.cursor()
        print('[提示]: 连接成功')

    def delete(self, params):
        """根据条件，删除表数据"""
        sql = "delete scott.dept_bak t where t.deptno = :deptno"
        self.cursor.execute(sql, params)
        self._conn.commit()
        print(f'[提示]: 删除成功，条件：{params}')


if __name__ == '__main__':
    test = Oracle('scott', '123456', '127.0.0.1:1521', 'orcl')

    # 测试
    test.delete({'deptno': 10})

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22

    2.5 调用存储过程

    import cx_Oracle


class Oracle(object):
    def __init__(self, user_name, password, host, instance):
        self._conn = cx_Oracle.connect(f'{user_name}/{password}@{host}/{instance}')
        self.cursor = self._conn.cursor()
        print('[提示]: 连接成功')

    def procedure(self, params):
        """存储过程"""
        sql = "begin pkg_test.pro_test(:i_deptno, :i_dname, :i_loc); end;"
        self.cursor.execute(sql, params)
        self._conn.commit()
        print(f'[提示]: 调用存储过程成功，参数：{params}')


if __name__ == '__main__':
    test = Oracle('scott', '123456', '127.0.0.1:1521', 'orcl')

    # 测试
    test.procedure({'i_deptno': 60, 'i_dname': 'aaa', 'i_loc': 'bbb'})

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22

    pkg head：

    create or replace package pkg_test is
   procedure pro_test(i_deptno in number,
                      i_dname  in varchar2,
                      i_loc    in varchar2);
end pkg_test;

    • 1
    • 2
    • 3
    • 4
    • 5

    pkg body：

    create or replace package body pkg_test is

   procedure pro_test(i_deptno in number,
                      i_dname  in varchar2,
                      i_loc    in varchar2) is
   begin
      execute immediate 'insert into scott.dept_bak(deptno, dname, loc) values(:b1, :b2, :b3)'
         using i_deptno, i_dname, i_loc;
   
      commit;
   end pro_test;
end pkg_test;

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
  • 相关阅读:
    Lambda表达式：一篇文章带你通透
    【C++】面向对象编程示例 ( 案例需求 | Visual Studio 创建类 | 类的声明 | 类的实现 | 类的调用 )
    队列--环形数组实现
    PNAS：睡眠的fMRI频谱特征
    word目录怎么自动生成？用这个方法，快速自动生成
    centos-apache-简易搭建静态网页服务器-总结
    896. 最长上升子序列 II 线性dp (优化版 nlogn 贪心+二分)
    Chat Towards Data Science｜如何用个人数据知识库构建 RAG 聊天机器人？
    【Web前端】HTML详解(上篇)
    docker常用基本命令
  • 原文地址：https://blog.csdn.net/qq_34745941/article/details/132729732