用户在创建好数据仓库集群后使用psycopg2第三方库连接到集群，则可以使用Python访问DWS ，并进行数据表的各类操作。

连接集群前的准备

DWS 集群已绑定弹性IP。
已获取DWS 集群的数据库管理员用户名和密码。


                注意
                由于MD5算法已经被证实存在碰撞可能，已严禁将之用于密码校验算法。当前DWS 采用默认安全设计，默认禁止MD5算法的密码校验，可能导致开源客户端无法正常连接的问题。建议先检查一下数据库参数password_encryption_type参数是否为1，如果取值不为1，需要修改，修改方法参见《用户指南》的“[修改数据库参数](https://www.ctyun.cn/document/10014061/10047788)”章节；然后修改一次准备使用的数据库用户的密码。


                说明
                当前DWS出于安全考虑，已经默认不再使用MD5存储密码摘要了，这将导致使用开源驱动或者客户端无法正常连接数据库。需要您调整一下密码策略后再创建一个新用户或者对老用户做一次密码修改，方可使用开源协议中的MD5认证算法。

数据库中是不会存储用户的密码原文，而是存储密码的HASH摘要，在密码校验时与客户端发来的密码摘要进行比对（中间会有加盐操作）。故当您改变了密码算法策略时，数据库也是无法还原您的密码，再生成新的HASH算法的摘要值的。必须您手动修改一次密码或者创建一个新用户，这时新的密码将会采用您设置的HASH算法进行摘要存储，用于下次连接认证。

已获取DWS 集群的公网访问地址，含IP地址和端口。具体请参见 [获取集群连接地址](https://www.ctyun.cn/document/10014061/10047782)。
已安装psycopg2第三方库。下载地址：[https://pypi.org/project/psycopg2/](https://pypi.org/project/psycopg2/ " ")，安装部署操作请参见：[https://www.psycopg.org/install/](https://www.psycopg.org/install/ " ")。


                说明
                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动态库。

使用约束

由于psycopg2是基于PostgreSQL的客户端接口，它的功能DWS并不能完全支持。具体支持情况请见下表。


                说明
                以下接口支持情况是基于Python 3.8.5及psycopg 2.9.1版本。

DWS对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	fileno()	Y	-
connections	isexecuting()	Y	-
connections	Interoperation with other C API modules	pgconn_ptr	Y	-
connections	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')  # 数据库用户密码

4.执行以下命令，使用psycopg第三方库连接集群。

python python_dws.py

在Windows环境使用psycopg2第三方库连接集群

1.在Windows系统中，单击“开始”按钮，在搜索框中，键入 cmd ，然后在结果列表中单击“cmd.exe”打开命令提示符窗口。

2.在命令提示符窗口中，执行以下命令创建python_dws.py文件。

type nul> python_dws.py

请复制粘贴以下内容放入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='postgresgaussdb',  # 需要连接的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文件中的集群公网访问地址、集群端口号、数据库名称、数据库用户名、数据库密码。

conn = psycopg2.connect(host='10.154.70.231', 
                                port='8000', 
                                database='gaussdb',  # 需要连接的database 
                                user='dbadmin', 
                                password='password')  # 数据库用户密码

4.在命令提示符窗口中，执行以下命令，使用psycopg第三方库连接集群。

python python_dws.py

psycopg2连接集群不支持CN Retry特性的问题说明

DWS支持在SQL语句执行出错时的自动重试功能（简称CN Retry）。CN Retry对于客户端和驱动发送的SQL语句在执行失败时可以自动识别错误类型，并进行重试。但使用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()

使用异步连接方式主动开启事务，异步连接介绍具体请参见pyscopg官网。

#!/usr/bin/env python3 
# _*_ encoding=utf-8 _*_ 
  
import psycopg2 
import select 
  
# psycopg2官方提供的异步连接方式时的wait函数 

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_sync(): 
    # 创建连接 
    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()

应用商城

合作伙伴

开发者

支持与服务

了解天翼云

查看所有产品

数据仓库服务

数据仓库服务

连接集群前的准备

使用约束

在Linux环境使用psycopg2第三方库连接集群

在Windows环境使用psycopg2第三方库连接集群

psycopg2连接集群不支持CN Retry特性的问题说明

报错原因：

解决方案：

活动

应用商城

合作伙伴

开发者

支持与服务

了解天翼云

查看所有产品

数据仓库服务

数据仓库服务

连接集群前的准备

使用约束

在Linux环境使用psycopg2第三方库连接集群

在Windows环境使用psycopg2第三方库连接集群

psycopg2连接集群不支持CN Retry特性的问题说明

报错原因：

解决方案：