使用cx_Oracle可连接Oracle，但SQLAlchemy连接失败求助

问题场景 #

使用cx_Oracle可以正常连接Oracle数据库，但通过SQLAlchemy创建引擎时触发错误。

原代码 #

import cx_Oracle
import sqlalchemy
import pandas as pd
from sqlalchemy import create_engine

hostname = r'myhost'
port = '0000'
sid = 'ssss'
username = 'myusername'
password = 'mypass'

dsn_tns = cx_Oracle.makedsn(hostname, port, sid) 
conn = cx_Oracle.connect(username, password=password, dsn=dsn_tns) 
conn.connect()
print('Connected to oracle using cx_oracle')
engine = create_engine(conn, connect_args={ "encoding": "UTF-8", "nencoding": "UTF-8"})
engine.connect()
print('Connected to sql alchemy')

错误输出 #

Connected to oracle using cx_oracle 
Traceback (most recent call last):
File "D:\Project\Python Script to connect to oracle sql\sqlalchemy_connect.py", line 16, in 
engine = create_engine(conn, connect_args={ "encoding": "UTF-8", "nencoding": "UTF-8"})   
File "", line 2, in create_engine
File "C:\Users\mikim\AppData\Local\Programs\Python\Python310\lib\site-packages\sqlalchemy\util\deprecations.py", line 277, in warned
return fn(*args, **kwargs)  # type: ignore[no-any-return]   
File "C:\Users\mikim\AppData\Local\Programs\Python\Python310\lib\site-packages\sqlalchemy\engine\create.py", line 549, in create_engine
u, plugins, kwargs = u._instantiate_plugins(kwargs) 
AttributeError: 'cx_Oracle.Connection' object has no attribute '_instantiate_plugins'

错误原因 #

create_engine()的第一个参数需要传入SQLAlchemy格式的连接URL，而非已经通过cx_Oracle创建好的Connection对象。SQLAlchemy会自行处理连接的创建与管理，不需要提前用cx_Oracle建立连接。

修复后的代码 #

方式1：直接构建连接URL #

import cx_Oracle
import sqlalchemy
import pandas as pd
from sqlalchemy import create_engine

hostname = 'myhost'
port = '0000'
sid = 'ssss'
username = 'myusername'
password = 'mypass'

# 构建SQLAlchemy Oracle连接URL，格式：oracle+cx_oracle://用户名:密码@主机:端口/?sid=SID
engine = create_engine(
    f"oracle+cx_oracle://{username}:{password}@{hostname}:{port}/?sid={sid}",
    connect_args={"encoding": "UTF-8", "nencoding": "UTF-8"}
)

try:
    with engine.connect() as conn:
        print('Connected to oracle using SQLAlchemy')
except Exception as e:
    print(f"连接失败: {str(e)}")

方式2：基于cx_Oracle的DSN构建URL #

import cx_Oracle
import sqlalchemy
from sqlalchemy import create_engine

hostname = 'myhost'
port = '0000'
sid = 'ssss'
username = 'myusername'
password = 'mypass'

# 用cx_Oracle生成DSN，再传入SQLAlchemy连接URL
dsn_tns = cx_Oracle.makedsn(hostname, port, sid)
engine = create_engine(
    f"oracle+cx_oracle://{username}:{password}@{dsn_tns}",
    connect_args={"encoding": "UTF-8", "nencoding": "UTF-8"}
)

try:
    with engine.connect() as conn:
        print('Connected to oracle using SQLAlchemy')
except Exception as e:
    print(f"连接失败: {str(e)}")

注意事项 #

• 若密码包含特殊字符（如@、#、$），需用urllib.parse.quote_plus()对密码进行URL编码，避免解析错误：

  from urllib.parse import quote_plus
  encoded_password = quote_plus(password)
  engine = create_engine(f"oracle+cx_oracle://{username}:{encoded_password}@{hostname}:{port}/?sid={sid}")
  

• 原代码中conn.connect()属于冗余操作，cx_Oracle.connect()已经完成连接创建与打开。
• 确保已安装依赖：pip install sqlalchemy cx-oracle（若使用SQLAlchemy 2.0+，也可替换为oracledb）

内容的提问来源于stack exchange，提问作者Michael