首页 计算机科学

# 使用 Python 连接 Oracle 数据库

  • 同步方式
    • python-oracledb
    • pyodbc
    • SQLAlchemy
    • SQLModel （对 SQLAlchemy 和 PyDantic 的封装）
  • 异步方式
    • aioodbc （异步的 pyodbc ）

# oracledb

python-oracledb 是 cx_Oracle 改名后的最新版，显然使用 oracledb 名称更简洁明确。

首先要下载解压 Oracle Instant Client 并配置环境变量，以开启 oracledb 的 Thick 模式。

首先安装包：

pip install oracledb

愉快地导包：

import oracledb

代码开启 Thick 模式：

oracledb.init_oracle_client()

代码创建数据库连接实例：

conn = oracledb.connect(
    user="scott",
    password="scott",
    host="localhost",
    port=1521,
    service_name="orcl",
)

这里首先解锁了 Oracle 数据库中的 scott 用户，并且设置密码为 scott ，也可选择其它已解锁用户。

尝试打印连接数据库软件的版本：

print('Database version:', conn.version)

创建游标对象以操纵数据库（传递 SQL 语句）：

cursor = conn.cursor()

查询语句：

query = "SELECT SYSDATE FROM DUAL"

执行 SQL 语句：

cursor.execute(query)

获取并打印语句查询的结果：

res = cursor.fetchone()
print(res)
(datetime.datetime(2022, 9, 13, 20, 6, 18),)

在不使用 with 上下文管理时，养成程序结束前关闭游标对象和数据库连接实例的习惯：

cursor.close()
conn.close()

完整示例（使用 with 上下文管理简化代码）：

import oracledb
oracledb.init_oracle_client()  # Thick 模式
with oracledb.connect(
    user="scott",
    password="scott",
    host="localhost",
    port=1521,
    service_name="orcl",
) as conn:
    # 打印数据库版本
    print("Database version:", conn.version)
    with conn.cursor() as cursor:
        # SQL 语句在这里不用加分号结束
        query = "SELECT SYSDATE FROM DUAL"
        cursor.execute(query)
        res = cursor.fetchone()
        print(res)
# pyodbc
首先安装包：
pip install pyodbc
愉快地导包：
import pyodbc
代码创建数据库连接实例：
conn = pyodbc.connect('''
    Driver={Orcale in OraDb11g_home1};
    Server=localhost;
    Database=orcl;
    Port=1521;
    UID=scott;
    PWD=scott;
''')
配置字段字母大小写不区分，但值要区分。
创建游标对象以操纵数据库（传递 SQL 语句）：
cursor = conn.cursor()
查询语句：
query = "SELECT SYSDATE FROM DUAL"
执行 SQL 语句：
cursor.execute(query)
获取并打印语句查询的结果：
res = cursor.fetchone()
print(res)
(datetime.datetime(2022, 9, 13, 20, 6, 18),)

同样，在不使用 with 上下文管理时，养成程序结束前关闭游标对象和数据库连接实例的习惯：
cursor.close()
conn.close()
完整示例（使用 with 上下文管理简化代码）：
import pyodbc
# print(pyodbc.drivers())
with pyodbc.connect('''
    Driver={Oracle in OraDb11g_home1};
    Server=localhost;
    Database=orcl;
    Port=1521;
    UID=scott;
    PWD=scott;
''') as conn:
    with conn.cursor() as cur:
        query = "SELECT SYSDATE FROM DUAL"
        cur.execute(query)
        res = cur.fetchone()
        print(res)
# SQLAlchemy
SQLAlchemy v1 使用的 cx_Oracle ，而 v2 版才支持 python-oracledb [1]，所以目前需要做一些模块调用的处理[2]：
import sys
import oracledb
oracledb.version = "8.3.0"
sys.modules["cx_Oracle"] = oracledb
import cx_Oracle
尽管对 import cx_Oracle 编辑器会显示红线错误，但解释器是识别的。
本质上用的 python-orcaledb ，别忘了开启 Thick 模式
oracledb.init_oracle_client()
对于中文环境，最好设置 UTF8 编码：
import os
os.environ['NLS_LANG'] = 'SIMPLIFIED CHINESE_CHINA.UTF8'
首先安装包：
pip install SQLAlchemy oracledb
愉快地导包[3]：
from sqlalchemy import create_engine
自定义数据源名称 dsn ：
dsn = "oracle+cx_oracle://scott:scott@localhost:1521/orcl"
# v2
# dsn = "oracle+oracledb://scott:scott@localhost:1521/orcl"
创建引擎：
engine = create_engine(dsn)
代码创建数据库连接实例：
conn = engine.connect()
查询语句：
query = "SELECT SYSDATE FROM DUAL"
获取并打印语句查询的结果：
cursor = conn.execute(query)
print(cursor.fetchone())
(datetime.datetime(2022, 9, 13, 20, 6, 18),)

同样，在不使用 with 上下文管理时，养成程序结束前关闭游标对象（可选）和数据库连接实例的习惯：
cursor.close()  # 通过 with 语句看出可以不用手动关闭
conn.close()
使用 Pandas 操纵数据库：
首先安装包：
pip install pandas
import pandas as pd
df = pd.read_sql(query, con=engine)
print(df['sysdate'].values.tolist())
[16630707780000000000]

完整示例（使用 with 上下文管理简化代码）：
import os
os.environ['NLS_LANG'] = 'SIMPLIFIED CHINESE_CHINA.UTF8'
import sys
import oracledb
oracledb.init_oracle_client()  # Thick mode
oracledb.version = "8.3.0"
sys.modules["cx_Oracle"] = oracledb
import cx_Oracle
from sqlalchemy import create_engine
dsn = "oracle+cx_oracle://scott:scott@localhost:1521/ORCL"
engine = create_engine(dsn)
query = "SELECT SYSDATE FROM DUAL"
with engine.connect() as conn:
    cursor conn.execute(query)
    print(cursor.fetchone())
import pandas as pd
df = pd.read_sql(query, con=engine)
print(df['sysdate'].values.tolist())
# SQLModel[4]
SQLModel 对 SQLAlchemy 和 PyDantic 作了更好的封装，使用面向对象编程和类型注解（类型提示），更方便地代码补全
首先安装包：
pip install sqlmodel oracledb
同 SQLAlchemy 一样提前导入一些包：
from datetime import datetime
import os
os.environ['NLS_LANG'] = 'SIMPLIFIED CHINESE_CHINA.UTF8'
import sys
import oracledb
oracledb.version = "8.3.0"
sys.modules["cx_Oracle"] = oracledb
import cx_Oracle
oracledb.init_oracle_client()  # Thick mode
愉快地导包：
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine, select
仿照所操纵表（这里是 EMP 表）的表结构（Schema）设计类，方便与数据库的表进行映射：
class Emp(SQLModel, table=True):
    empno: Optional[int] = Field(default=None, primary_key=True)  # EMPNO 字段作为主键且不能为空值，但是 Python 的变量必须传入一个值，不然报错
    ename: str
    job: str
    mgr: int
    hiredate: datetime
    sal: int
    comm: Optional[int] = Field(default=None)
自定义数据源名称 dsn ：
dsn = "oracle+cx_oracle://scott:scott@localhost:1521/orcl"
# v2
# dsn = "oracle+oracledb://scott:scott@localhost:1521/orcl"
创建引擎：
engine = create_engine(dsn)
代码创建数据库连接会话：
session = Session(engine)
查询语句：
query = select(Emp)
获取并打印语句查询的结果：
emp = session.exec(query).fetchall()
print(emp)
[Emp(mgr=7902, ename='SMITH', job='CLERK', sal=800, empno=7369, hiredate=datetime.datetime(1980, 12, 17, 0, 0), comm=None), Emp(mgr=7698, ename='ALLEN', job='SALESMAN', sal=1600, empno=7499, hiredate=datetime.datetime(1981, 2, 20, 0, 0), comm=300), Emp(mgr=7698, ename='WARD', job='SALESMAN', sal=1250, empno=7521, hiredate=datetime.datetime(1981, 2, 22, 0, 0), comm=500), Emp(mgr=7839, ename='JONES', job='MANAGER', sal=2975, empno=7566, hiredate=datetime.datetime(1981, 4, 2, 0, 0), comm=None), Emp(mgr=7698, ename='MARTIN', job='SALESMAN', sal=1250, empno=7654, hiredate=datetime.datetime(1981, 9, 28, 0, 0), comm=1400), Emp(mgr=7839, ename='BLAKE', job='MANAGER', sal=2850, empno=7698, hiredate=datetime.datetime(1981, 5, 1, 0, 0), comm=None), Emp(mgr=7839, ename='CLARK', job='MANAGER', sal=2450, empno=7782, hiredate=datetime.datetime(1981, 6, 9, 0, 0), comm=None), Emp(mgr=7566, ename='SCOTT', job='ANALYST', sal=3000, empno=7788, hiredate=datetime.datetime(1987, 4, 19, 0, 0), comm=None), Emp(mgr=None, ename='KING', job='PRESIDENT', sal=5000, empno=7839, hiredate=datetime.datetime(1981, 11, 17, 0, 0), comm=None), Emp(mgr=7698, ename='TURNER', job='SALESMAN', sal=1500, empno=7844, hiredate=datetime.datetime(1981, 9, 8, 0, 0), comm=0), Emp(mgr=7788, ename='ADAMS', job='CLERK', sal=1100, empno=7876, hiredate=datetime.datetime(1987, 5, 23, 0, 0), comm=None), Emp(mgr=7698, ename='JAMES', job='CLERK', sal=950, empno=7900, hiredate=datetime.datetime(1981, 12, 3, 0, 0), comm=None), Emp(mgr=7566, ename='FORD', job='ANALYST', sal=3000, empno=7902, hiredate=datetime.datetime(1981, 12, 3, 0, 0), comm=None), Emp(mgr=7782, ename='MILLER', job='CLERK', sal=1300, empno=7934, hiredate=datetime.datetime(1982, 1, 23, 0, 0), comm=None)]

同样，在不使用 with 上下文管理时，养成程序结束前关闭数据库连接会话的习惯：
session.close()
完整示例（使用 with 上下文管理简化代码）：
from datetime import datetime
import os
os.environ['NLS_LANG'] = 'SIMPLIFIED CHINESE_CHINA.UTF8'
import sys
import oracledb
oracledb.init_oracle_client()  # Thick mode
oracledb.version = "8.3.0"
sys.modules["cx_Oracle"] = oracledb
import cx_Oracle
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Emp(SQLModel, table=True):
    empno: Optional[int] = Field(default=None, primary_key=True)
    ename: str
    job: str
    mgr: int
    hiredate: datetime
    sal: int
    comm: Optional[int] = Field(default=None)
dsn = "oracle+cx_oracle://scott:scott@localhost:1521/orcl"
engine = create_engine(dsn)
with Session(engine) as session:
    statement = select(Emp)
    emp = session.exec(statement).fetchall()
    print(emp)
# aioobdc[5]
需要 Python 3.6+
首先安装包：
pip install aioodbc
愉快地导包：
import asyncio
import aioodbc
自定义数据源名称 dsn ：
dsn = '''
    Driver={Oracle in OraDb11g_home1};
    Server=localhost;
    Database=orcl;
    Port=1521;
    UID=scott;
    PWD=scott;
'''
定义异步函数：
async def test():
    conn = await aioodbc.connect(dsn=dsn)
    cursor = await conn.cursor()
    await cursor.execute("SELECT SYSDATE FROM DUAL")
    res = await cursor.fetchone()
    print(res)
    
    # 同样，在不使用 `with` 上下文管理时，养成程序结束前关闭游标对象和数据库连接实例的习惯
    await cursor.close()
    await conn.close()
调用执行异步函数：
loop = asyncio.get_event_loop()
loop.run_until_complete(test())
对于 Python 3.7+ ：
asyncio.run(test())
获取并打印语句查询的结果：
(datetime.datetime(2022, 9, 13, 20, 6, 18),)

完整示例（使用 with 上下文管理简化代码）：
import asyncio
import aioodbc
dsn = '''
    Driver={Oracle in OraDb11g_home1};
    Server=localhost;
    Database=orcl;
    Port=1521;
    UID=scott;
    PWD=scott;
'''
async def test():
    async with aioodbc.connect(dsn=dsn) as conn:
        async with conn.cursor() as cursor:
            await cursor.execute("SELECT SYSDATE FROM DUAL")
            res = await cursor.fetchone()
            print(res)
asyncio.run(test())
# 使用 Golang 连接 Oracle 数据库[6]
  1. How to use oracledb with sqlalchemy 1.4 ↩︎
  2. Python Frameworks, SQL Generators, and ORMs ↩︎
  3. Oracle - SQLAlchemy 1.4 Documentation ↩︎
  4. SQLModel ↩︎
  5. aioodbc GitHub ↩︎
  6. go-ora ↩︎