# 使用Python第三方库psycopg2连接DWS集群
用户在创建好数据仓库集群后，可以通过Python第三方库psycopg2连接DWS集群，使用Python访问DWS，并对数据表执行建表、插入数据、查询数据、更新数据、删除数据等各类操作。
#### 应用场景
psycopg2适用于以下业务场景：
- 使用Python开发的Web应用或数据处理脚本需要直接操作DWS中的数据。
- 已有的Python应用需要对接DWS，并复用psycopg2生态的成熟能力。
- 需要通过Python对数据表执行建表、增删改查等操作。
**示例**：例如，Web应用的后台服务通过psycopg2查询DWS中的业务数据并返回给前端，或将用户操作记录批量写入DWS。
#### 连接集群前的准备
连接集群前，请确认以下准备工作已完成：
- 已创建DWS集群，且集群已绑定弹性IP。
- 已获取DWS集群的数据库管理员用户名和密码。
- 由于MD5算法已被证实存在碰撞风险，DWS默认禁止使用MD5密码校验，若使用开源客户端连接时提示认证失败，通常是该客户端仍采用MD5认证所致。请按以下步骤处理：
1. 联系技术支持人员检查数据库参数password_encryption_type参数是否为1，如果取值不为1，请修改该参数。
2. 修改一次准备使用的数据库用户的密码，或新建一个数据库用户。
- 数据库不存储您的密码原文，仅存储密码的哈希（HASH）摘要，在密码校验时与客户端发来的密码摘要进行比对（比对过程包含加盐操作）。因此，当密码算法策略变更后，数据库无法基于原密码自动生成新算法的摘要，必须手动修改一次密码或新建用户，新的密码才会按新算法生成摘要，用于下次连接认证。
- 已获取DWS集群的公网访问地址（含IP地址和端口）。具体请参见[获取DWS集群连接地址](https://support.huaweicloud.com/mgtg-dws/dws_01_0033.html)。
#### 安装psycopg2
psycopg2下载地址：，安装部署操作请参见：，各操作系统的安装方式如下：
- CentOS、Redhat等操作系统中，使用yum命令安装：
```
yum install python-psycopg2
```
- psycopg2的使用依赖PostgreSQL的libpq动态库（32位的psycopg2对应32位的libpq，64位的psycopg2对应64位的libpq），Linux环境可借助yum命令解决依赖。在Windows系统使用psycopg2需先安装libpq，主要有以下两种方式：
- 安装PostgreSQL，并将libpq、SSL、crypto动态库所在目录配置到环境变量PATH中。
- 安装psqlodbc，使用PostgreSQL ODBC驱动自带的libpq、SSL、crypto动态库。
#### 版本说明
由于DWS集群、Python、psycopg2的版本较多，下方表格仅列举出当前主流版本的支持情况。
表1DWS版本支持详情
| psycopg2版本 | Python版本 | DWS集群版本 |
|:---|:---|:---|
| 2.7.x | 3.8.x | 8.1.3及以上 |
| 2.7.x | 3.9.x | 8.1.3及以上 |
| 2.8.x | 3.8.x | 8.1.3及以上 |
| 2.8.x | 3.9.x | 8.1.3及以上 |
| 2.9.x | 3.8.x | 8.1.3及以上 |
| 2.9.x | 3.9.x | 8.1.3及以上 |
#### 使用约束
由于psycopg2是基于PostgreSQL的客户端接口，它的功能DWS并不能完全支持。具体支持情况请参见下[表2]。以下接口支持情况是基于Python 3.8.5及psycopg 2.9.1版本。
表2DWS对psycopg2主要接口支持情况
| 类名 | 功能描述 | 函数/成员变量 | 支持 | 备注 |
|:---|:---|:---|:---|:---|
| connections | basic | cursor(*name=None* , *cursor_factory=None* , *scrollable=None* , *withhold=False*) | Y | - |
| connections | basic | commit() | Y | - |
| connections | basic | rollback() | Y | - |
| connections | basic | close() | Y | - |
| connections | Two-phase commit support methods | xid(*format_id* , *gtrid* , *bqual*) | Y | - |
| connections | Two-phase commit support methods | tpc_begin(*xid*) | Y | - |
| connections | Two-phase commit support methods | tpc_prepare() | N | 内核不支持显式prepare transaction。 |
| connections | Two-phase commit support methods | tpc_commit(\[*xid*\]) | Y | - |
| connections | Two-phase commit support methods | tpc_rollback(\[*xid*\]) | Y | - |
| connections | Two-phase commit support methods | tpc_recover() | Y | - |
| connections | Two-phase commit support methods | closed | Y | - |
| connections | Two-phase commit support methods | cancel() | Y | - |
| connections | Two-phase commit support methods | reset() | N | 不支持DISCARD ALL。 |
| connections | Two-phase commit support methods | dsn | Y | - |
| connections | Transaction control methods and attributes. | set_session(*isolation_level=None* , *readonly=None* , *deferrable=None* , *autocommit=None*) | Y | 数据库不支持session中设置default_transaction_read_only。 |
| connections | Transaction control methods and attributes. | autocommit | Y | - |
| connections | Transaction control methods and attributes. | isolation_level | Y | - |
| connections | Transaction control methods and attributes. | readonly | N | 数据库不支持session中设置default_transaction_read_only。 |
| connections | Transaction control methods and attributes. | deferrable | Y | - |
| connections | Transaction control methods and attributes. | set_isolation_level(*level*) | Y | - |
| connections | Transaction control methods and attributes. | encoding | Y | - |
| connections | Transaction control methods and attributes. | set_client_encoding(enc) | Y | - |
| connections | Transaction control methods and attributes. | notices | N | 数据库不支持listen/notify。 |
| connections | Transaction control methods and attributes. | notifies | Y | - |
| connections | Transaction control methods and attributes. | cursor_factory | Y | - |
| connections | Transaction control methods and attributes. | info | Y | - |
| connections | Transaction control methods and attributes. | status | Y | - |
| connections | Transaction control methods and attributes. | lobject | N | 数据库不支持大对象相关操作。 |
| connections | Methods related to asynchronous support | poll() | Y | - |
| connections | Methods related to asynchronous support | fileno() | Y | - |
| connections | Methods related to asynchronous support | isexecuting() | Y | - |
| connections | Interoperation with other C API modules | pgconn_ptr | Y | - |
| connections | Interoperation with other C API modules | get_native_connection() | Y | - |
| connections | informative methods of the native connection | get_transaction_status() | Y | - |
| connections | informative methods of the native connection | protocol_version | Y | - |
| connections | informative methods of the native connection | server_version | Y | - |
| connections | informative methods of the native connection | get_backend_pid() | Y | 获取到的不是后台的pid，是逻辑连接的id号。 |
| connections | informative methods of the native connection | get_parameter_status(parameter) | Y | - |
| connections | informative methods of the native connection | get_dsn_parameters() | Y | - |
| cursor | basic | description | Y | - |
| cursor | basic | close() | Y | - |
| cursor | basic | closed | Y | - |
| cursor | basic | connection | Y | - |
| cursor | basic | name | Y | - |
| cursor | basic | scrollable | N | 数据库不支持SCROLL CURSOR。 |
| cursor | basic | withhold | N | withhold cursor在commit前需要关闭。 |
| cursor | Commands execution methods | execute(*query* , *vars=None*) | Y | - |
| cursor | Commands execution methods | executemany(*query* , *vars_list*) | Y | - |
| cursor | Commands execution methods | callproc(*procname* \[, *parameters*\]) | Y | - |
| cursor | Commands execution methods | mogrify(*operation* \[, *parameters*\]) | Y | - |
| cursor | Commands execution methods | setinputsizes(*sizes*) | Y | - |
| cursor | Commands execution methods | fetchone() | Y | - |
| cursor | Commands execution methods | fetchmany(\[*size=cursor.arraysize*\]) | Y | - |
| cursor | Commands execution methods | fetchall() | Y | - |
| cursor | Commands execution methods | scroll(*value* \[, *mode='relative'*\]) | N | 数据库不支持SCROLL CURSOR。 |
| cursor | Commands execution methods | arraysize | Y | - |
| cursor | Commands execution methods | itersize | Y | - |
| cursor | Commands execution methods | rowcount | Y | - |
| cursor | Commands execution methods | rownumber | Y | - |
| cursor | Commands execution methods | lastrowid | Y | - |
| cursor | Commands execution methods | query | Y | - |
| cursor | Commands execution methods | statusmessage | Y | - |
| cursor | Commands execution methods | cast(*oid* , *s*) | Y | - |
| cursor | Commands execution methods | tzinfo_factory | Y | - |
| cursor | Commands execution methods | nextset() | Y | - |
| cursor | Commands execution methods | setoutputsize(*size* \[, *column*\]) | Y | - |
| cursor | COPY-related methods | copy_from(*file* , *table* , *sep='\\\\t'* , *null='\\\\\\\\N'* , *size=8192* , *columns=None*) | Y | - |
| cursor | COPY-related methods | copy_to(*file* , *table* , *sep='\\\\t'* , *null='\\\\\\\\N'* , *columns=None*) | Y | - |
| cursor | COPY-related methods | copy_expert(*sql* , *file* , *size=8192*) | Y | - |
| cursor | Interoperation with other C API modules | pgresult_ptr | Y | - |
#### 在Linux环境使用psycopg2第三方库连接集群
1. 以**root**用户登录Linux环境。
2. 执行以下命令创建**python_dws.py** 文件。
```
vi python_dws.py
```
请复制粘贴以下内容放入**python_dws.py**文件中：
```
#!/usr/bin/python
# -*- coding: UTF-8 -*-
from __future__ import print_function
import psycopg2
def create_table(connection):
print("Begin to create table")
try:
cursor = connection.cursor()
cursor.execute("drop table if exists test;"
"create table test(id int, name text);")
connection.commit()
except psycopg2.ProgrammingError as e:
print(e)
else:
print("Table created successfully")
cursor.close()
def insert_data(connection):
print("Begin to insert data")
try:
cursor = connection.cursor()
cursor.execute("insert into test values(1,'number1');")
cursor.execute("insert into test values(2,'number2');")
cursor.execute("insert into test values(3,'number3');")
connection.commit()
except psycopg2.ProgrammingError as e:
print(e)
else:
print("Insert data successfully")
cursor.close()
def update_data(connection):
print("Begin to update data")
try:
cursor = connection.cursor()
cursor.execute("update test set name = 'numberupdated' where id=1;")
connection.commit()
print("Total number of rows updated :", cursor.rowcount)
cursor.execute("select * from test order by 1;")
rows = cursor.fetchall()
for row in rows:
print("id = ", row[0])
print("name = ", row[1], "\n")
except psycopg2.ProgrammingError as e:
print(e)
else:
print("After Update, Operation done successfully")
def delete_data(connection):
print("Begin to delete data")
try:
cursor = connection.cursor()
cursor.execute("delete from test where id=3;")
connection.commit()
print("Total number of rows deleted :", cursor.rowcount)
cursor.execute("select * from test order by 1;")
rows = cursor.fetchall()
for row in rows:
print("id = ", row[0])
print("name = ", row[1], "\n")
except psycopg2.ProgrammingError as e:
print(e)
else:
print("After Delete,Operation done successfully")
def select_data(connection):
print("Begin to select data")
try:
cursor = connection.cursor()
cursor.execute("select * from test order by 1;")
rows = cursor.fetchall()
for row in rows:
print("id = ", row[0])
print("name = ", row[1], "\n")
except psycopg2.ProgrammingError as e:
print(e)
print("select failed")
else:
print("Operation done successfully")
cursor.close()
if __name__ == '__main__':
try:
conn = psycopg2.connect(host='10.154.70.231',
port='8000',
database='gaussdb', # 需要连接的database
user='dbadmin',
password='password') # 数据库用户密码
except psycopg2.DatabaseError as ex:
print(ex)
print("Connect database failed")
else:
print("Opened database successfully")
create_table(conn)
insert_data(conn)
select_data(conn)
update_data(conn)
delete_data(conn)
conn.close()
```
3. 按照实际集群信息，修改**python_dws.py** 文件中的集群公网访问地址、集群端口号、数据库名称、数据库用户名、数据库密码。
psycopg2接口不提供重试连接的能力，您需要在业务代码中实现重试处理。
```
conn = psycopg2.connect(host='10.154.70.231',
port='8000',
database='gaussdb', # 需要连接的database
user='dbadmin',
password='password') # 数据库用户密码
```
表3连接集群参数说明
| 参数 | 说明 |
|:---|:---|
| host | DWS集群的公网访问地址（弹性IP地址），请替换为实际地址，获取详情请参见[表4](https://support.huaweicloud.com/mgtg-dws/dws_01_0112.html#ZH-CN_TOPIC_0000002587065532__table112851640172712)。 |
| port | DWS集群的数据库端口号，默认**8000**。 |
| database | 需要连接的数据库名称，默认数据库为**gaussdb**。 |
| user | 数据库用户名，默认数据库管理员为**dbadmin**。 |
| password | 数据库用户密码，请替换为实际密码。 |
4. 执行以下命令，使用psycopg第三方库连接集群。
```
python python_dws.py
```
5. 执行命令后，若输出成功连接数据库的信息，并依次出现建表、插入数据、查询、更新、删除等操作成功的提示，则表明数据库连接正常，且各数据表操作均已成功执行。
#### 在Windows环境使用psycopg2第三方库连接集群
1. 在Windows系统中，单击**"开始** **"** ，在搜索框中，键入**cmd** ，然后在结果列表中单击**"cmd.exe"**打开命令提示符窗口。
2. 在命令提示符窗口中，执行以下命令创建**python_dws.py** 文件。
```
type nul> python_dws.py
```
请复制粘贴以下内容放入**python_dws.py**文件中：
```
#!/usr/bin/python
# -*- coding:UTF-8 -*-
from __future__ import print_function
import psycopg2
def create_table(connection):
print("Begin to create table")
try:
cursor = connection.cursor()
cursor.execute("drop table if exists test;"
"create table test(id int, name text);")
connection.commit()
except psycopg2.ProgrammingError as e:
print(e)
else:
print("Table created successfully")
cursor.close()
def insert_data(connection):
print("Begin to insert data")
try:
cursor = connection.cursor()
cursor.execute("insert into test values(1,'number1');")
cursor.execute("insert into test values(2,'number2');")
cursor.execute("insert into test values(3,'number3');")
connection.commit()
except psycopg2.ProgrammingError as e:
print(e)
else:
print("Insert data successfully")
cursor.close()
def update_data(connection):
print("Begin to update data")
try:
cursor = connection.cursor()
cursor.execute("update test set name = 'numberupdated' where id=1;")
connection.commit()
print("Total number of rows updated :", cursor.rowcount)
cursor.execute("select * from test order by 1;")
rows = cursor.fetchall()
for row in rows:
print("id = ", row[0])
print("name = ", row[1], "\n")
except psycopg2.ProgrammingError as e:
print(e)
else:
print("After Update, Operation done successfully")
def delete_data(connection):
print("Begin to delete data")
try:
cursor = connection.cursor()
cursor.execute("delete from test where id=3;")
connection.commit()
print("Total number of rows deleted :", cursor.rowcount)
cursor.execute("select * from test order by 1;")
rows = cursor.fetchall()
for row in rows:
print("id = ", row[0])
print("name = ", row[1], "\n")
except psycopg2.ProgrammingError as e:
print(e)
else:
print("After Delete,Operation done successfully")
def select_data(connection):
print("Begin to select data")
try:
cursor = connection.cursor()
cursor.execute("select * from test order by 1;")
rows = cursor.fetchall()
for row in rows:
print("id = ", row[0])
print("name = ", row[1], "\n")
except psycopg2.ProgrammingError as e:
print(e)
print("select failed")
else:
print("Operation done successfully")
cursor.close()
if __name__ == '__main__':
try:
conn = psycopg2.connect(host='10.154.70.231',
port='8000',
database='gaussdb', # 需要连接的database
user='dbadmin',
password='password') # 数据库用户密码
except psycopg2.DatabaseError as ex:
print(ex)
print("Connect database failed")
else:
print("Opened database successfully")
create_table(conn)
insert_data(conn)
select_data(conn)
update_data(conn)
delete_data(conn)
conn.close()
```
3. 按照实际集群信息，修改**python_dws.py** 文件中的集群公网访问地址、集群端口号、数据库名称、数据库用户名、数据库密码，参数说明请参见[表3]。
```
conn = psycopg2.connect(host='10.154.70.231',
port='8000',
database='gaussdb', # 需要连接的database
user='dbadmin',
password='password') # 数据库用户密码
```
4. 在命令提示符窗口中，执行以下命令，使用psycopg第三方库连接集群。
```
python python_dws.py
```
5. 执行命令后，若输出成功连接数据库的信息，并依次出现建表、插入数据、查询、更新、删除等操作成功的提示，则表明数据库连接正常，且各数据表操作均已成功执行。
#### psycopg2连接集群不支持CN Retry特性的问题说明
DWS支持在SQL语句执行出错时的自动重试功能（简称**CN Retry** ）。CN Retry对于客户端和驱动发送的SQL语句在执行失败时可以自动识别错误类型，并进行重试，详情请参见[SQL语句出错自动重试](https://support.huaweicloud.com/devg-dws/dws_04_0497.html)。但使用psycopg2默认连接方式创建的连接在语句执行失败时没有自动重试，会直接报错退出。如常见的主备切换场景下，未自动重试会报如下错误，但在自动重试期间完成主备切换，则会返回正确结果。
```
psycopg2.errors.ConnectionFailure: pooler: failed to create 1 connections, Error Message: remote node dn_6003_6004, detail: could not connect to server: Operation now in progress
```
**报错原因：**
1. psycopg2在发送SQL语句前先发送了BEGIN语句开启事务。
2. CN Retry不支持事务块中的语句是特性约束。
**解决方案：**
- 在同步方式连接时，可以通过主动结束驱动开启的事务。
```
cursor = conn.cursor()
# 增加end语句主动结束驱动开启的事务
cursor.execute("end; select * from test order by 1;")
rows = cursor.fetchall()
```
- 使用异步连接方式主动开启事务，异步连接介绍具体请参见psycopg官网：。
```
#!/usr/bin/env python3
# _*_ encoding=utf-8 _*_
import psycopg2
import select
# psycopg2官方提供的异步连接方式时的wait函数
# 详见https://www.psycopg.org/docs/advanced.html?highlight=async
def wait(conn):
while True:
state = conn.poll()
if state == psycopg2.extensions.POLL_OK:
break
elif state == psycopg2.extensions.POLL_WRITE:
select.select([], [conn.fileno()], [])
elif state == psycopg2.extensions.POLL_READ:
select.select([conn.fileno()], [], [])
else:
raise psycopg2.OperationalError("poll() returned %s" % state)
def psycopg2_cnretry_async():
# 创建连接
conn = psycopg2.connect(host='10.154.70.231',
port='8000',
database='gaussdb', # 需要连接的database
user='dbadmin',
password='password', # 数据库用户密码
async=1) # 使用异步方式连接
wait(conn)
# 执行查询
cursor = conn.cursor()
cursor.execute("select * from test order by 1;")
wait(conn)
rows = cursor.fetchall()
for row in rows:
print(row[0], row[1])
# 关闭连接
conn.close()
if __name__ == '__main__':
psycopg2_cnretry_async()
```